diff options
Diffstat (limited to 'src/backend')
| -rw-r--r-- | src/backend/catalog/heap.c | 386 | ||||
| -rw-r--r-- | src/backend/catalog/information_schema.sql | 71 | ||||
| -rw-r--r-- | src/backend/catalog/pg_constraint.c | 441 | ||||
| -rw-r--r-- | src/backend/commands/tablecmds.c | 1664 | ||||
| -rw-r--r-- | src/backend/optimizer/util/plancat.c | 2 | ||||
| -rw-r--r-- | src/backend/parser/gram.y | 19 | ||||
| -rw-r--r-- | src/backend/parser/parse_utilcmd.c | 288 | ||||
| -rw-r--r-- | src/backend/utils/adt/ruleutils.c | 22 | ||||
| -rw-r--r-- | src/backend/utils/cache/relcache.c | 42 |
9 files changed, 544 insertions, 2391 deletions
diff --git a/src/backend/catalog/heap.c b/src/backend/catalog/heap.c index 922ba79ac25..08b8362d64d 100644 --- a/src/backend/catalog/heap.c +++ b/src/backend/catalog/heap.c @@ -2164,54 +2164,6 @@ StoreRelCheck(Relation rel, const char *ccname, Node *expr, } /* - * Store a not-null constraint for the given relation - * - * The OID of the new constraint is returned. - */ -static Oid -StoreRelNotNull(Relation rel, const char *nnname, AttrNumber attnum, - bool is_validated, bool is_local, int inhcount, - bool is_no_inherit) -{ - Oid constrOid; - - constrOid = - CreateConstraintEntry(nnname, - RelationGetNamespace(rel), - CONSTRAINT_NOTNULL, - false, - false, - is_validated, - InvalidOid, - RelationGetRelid(rel), - &attnum, - 1, - 1, - InvalidOid, /* not a domain constraint */ - InvalidOid, /* no associated index */ - InvalidOid, /* Foreign key fields */ - NULL, - NULL, - NULL, - NULL, - 0, - ' ', - ' ', - NULL, - 0, - ' ', - NULL, /* not an exclusion constraint */ - NULL, - NULL, - is_local, - inhcount, - is_no_inherit, - false, /* conperiod */ - false); - return constrOid; -} - -/* * Store defaults and constraints (passed as a list of CookedConstraint). * * Each CookedConstraint struct is modified to store the new catalog tuple OID. @@ -2255,14 +2207,6 @@ StoreConstraints(Relation rel, List *cooked_constraints, bool is_internal) is_internal); numchecks++; break; - - case CONSTR_NOTNULL: - con->conoid = - StoreRelNotNull(rel, con->name, con->attnum, - !con->skip_validation, con->is_local, - con->inhcount, con->is_no_inherit); - break; - default: elog(ERROR, "unrecognized constraint type: %d", (int) con->contype); @@ -2320,8 +2264,6 @@ AddRelationNewConstraints(Relation rel, ParseNamespaceItem *nsitem; int numchecks; List *checknames; - List *nnnames; - ListCell *cell; Node *expr; CookedConstraint *cooked; @@ -2352,9 +2294,8 @@ AddRelationNewConstraints(Relation rel, /* * Process column default expressions. */ - foreach(cell, newColDefaults) + foreach_ptr(RawColumnDefault, colDef, newColDefaults) { - RawColumnDefault *colDef = (RawColumnDefault *) lfirst(cell); Form_pg_attribute atp = TupleDescAttr(rel->rd_att, colDef->attnum - 1); Oid defOid; @@ -2407,10 +2348,8 @@ AddRelationNewConstraints(Relation rel, */ numchecks = numoldchecks; checknames = NIL; - nnnames = NIL; - foreach(cell, newConstraints) + foreach_node(Constraint, cdef, newConstraints) { - Constraint *cdef = (Constraint *) lfirst(cell); Oid constrOid; if (cdef->contype == CONSTR_CHECK) @@ -2444,14 +2383,12 @@ AddRelationNewConstraints(Relation rel, */ if (cdef->conname != NULL) { - ListCell *cell2; - ccname = cdef->conname; /* Check against other new constraints */ /* Needed because we don't do CommandCounterIncrement in loop */ - foreach(cell2, checknames) + foreach_ptr(char, chkname, checknames) { - if (strcmp((char *) lfirst(cell2), ccname) == 0) + if (strcmp(chkname, ccname) == 0) ereport(ERROR, (errcode(ERRCODE_DUPLICATE_OBJECT), errmsg("check constraint \"%s\" already exists", @@ -2465,7 +2402,7 @@ AddRelationNewConstraints(Relation rel, * 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 ATAddCheckNNConstraint wants.) + * which is what ATAddCheckConstraint wants.) */ if (MergeWithExistingConstraint(rel, ccname, expr, allow_merge, is_local, @@ -2534,107 +2471,6 @@ AddRelationNewConstraints(Relation rel, cooked->is_no_inherit = cdef->is_no_inherit; cookedConstraints = lappend(cookedConstraints, cooked); } - else if (cdef->contype == CONSTR_NOTNULL) - { - CookedConstraint *nncooked; - AttrNumber colnum; - char *nnname; - int existing; - - /* Determine which column to modify */ - colnum = get_attnum(RelationGetRelid(rel), strVal(linitial(cdef->keys))); - if (colnum == InvalidAttrNumber) /* shouldn't happen */ - elog(ERROR, "cache lookup failed for attribute \"%s\" of relation %u", - strVal(linitial(cdef->keys)), RelationGetRelid(rel)); - - /* - * If the column already has an inheritable not-null constraint, - * we need only adjust its coninhcount and we're done. In certain - * cases (see below), if the constraint is there but marked NO - * INHERIT, then we mark it as no longer such and coninhcount - * updated, plus we must also recurse to the children (if any) to - * add the constraint there. - * - * We only allow the inheritability status to change during binary - * upgrade (where it's used to add the not-null constraints for - * children of tables with primary keys), or when we're recursing - * processing a table down an inheritance hierarchy; directly - * allowing a constraint to change from NO INHERIT to INHERIT - * during ALTER TABLE ADD CONSTRAINT would be far too surprising - * behavior. - */ - existing = AdjustNotNullInheritance1(RelationGetRelid(rel), colnum, - cdef->inhcount, cdef->is_no_inherit, - IsBinaryUpgrade || allow_merge); - if (existing == 1) - continue; /* all done! */ - else if (existing == -1) - { - List *children; - - children = find_inheritance_children(RelationGetRelid(rel), NoLock); - foreach_oid(childoid, children) - { - Relation childrel = table_open(childoid, NoLock); - - AddRelationNewConstraints(childrel, - NIL, - list_make1(copyObject(cdef)), - allow_merge, - is_local, - is_internal, - queryString); - /* these constraints are not in the return list -- good? */ - - table_close(childrel, NoLock); - } - - continue; - } - - /* - * If a constraint name is specified, check that it isn't already - * used. Otherwise, choose a non-conflicting one ourselves. - */ - if (cdef->conname) - { - if (ConstraintNameIsUsed(CONSTRAINT_RELATION, - RelationGetRelid(rel), - cdef->conname)) - ereport(ERROR, - errcode(ERRCODE_DUPLICATE_OBJECT), - errmsg("constraint \"%s\" for relation \"%s\" already exists", - cdef->conname, RelationGetRelationName(rel))); - nnname = cdef->conname; - } - else - nnname = ChooseConstraintName(RelationGetRelationName(rel), - strVal(linitial(cdef->keys)), - "not_null", - RelationGetNamespace(rel), - nnnames); - nnnames = lappend(nnnames, nnname); - - constrOid = - StoreRelNotNull(rel, nnname, colnum, - cdef->initially_valid, - cdef->inhcount == 0, - cdef->inhcount, - cdef->is_no_inherit); - - nncooked = (CookedConstraint *) palloc(sizeof(CookedConstraint)); - nncooked->contype = CONSTR_NOTNULL; - nncooked->conoid = constrOid; - nncooked->name = nnname; - nncooked->attnum = colnum; - nncooked->expr = NULL; - nncooked->skip_validation = cdef->skip_validation; - nncooked->is_local = is_local; - nncooked->inhcount = cdef->inhcount; - nncooked->is_no_inherit = cdef->is_no_inherit; - - cookedConstraints = lappend(cookedConstraints, nncooked); - } } /* @@ -2804,218 +2640,6 @@ MergeWithExistingConstraint(Relation rel, const char *ccname, Node *expr, return found; } -/* list_sort comparator to sort CookedConstraint by attnum */ -static int -list_cookedconstr_attnum_cmp(const ListCell *p1, const ListCell *p2) -{ - AttrNumber v1 = ((CookedConstraint *) lfirst(p1))->attnum; - AttrNumber v2 = ((CookedConstraint *) lfirst(p2))->attnum; - - return pg_cmp_s16(v1, v2); -} - -/* - * Create the not-null constraints when creating a new relation - * - * These come from two sources: the 'constraints' list (of Constraint) is - * specified directly by the user; the 'old_notnulls' list (of - * CookedConstraint) comes from inheritance. We create one constraint - * for each column, giving priority to user-specified ones, and setting - * inhcount according to how many parents cause each column to get a - * not-null constraint. If a user-specified name clashes with another - * user-specified name, an error is raised. - * - * Note that inherited constraints have two shapes: those coming from another - * not-null constraint in the parent, which have a name already, and those - * coming from a primary key in the parent, which don't. Any name specified - * in a parent is disregarded in case of a conflict. - * - * Returns a list of AttrNumber for columns that need to have the attnotnull - * flag set. - */ -List * -AddRelationNotNullConstraints(Relation rel, List *constraints, - List *old_notnulls) -{ - List *givennames; - List *nnnames; - List *nncols = NIL; - ListCell *lc; - - /* - * We track two lists of names: nnnames keeps all the constraint names, - * givennames tracks user-generated names. The distinction is important, - * because we must raise error for user-generated name conflicts, but for - * system-generated name conflicts we just generate another. - */ - nnnames = NIL; - givennames = NIL; - - /* - * First, create all not-null constraints that are directly specified by - * the user. Note that inheritance might have given us another source for - * each, so we must scan the old_notnulls list and increment inhcount for - * each element with identical attnum. We delete from there any element - * that we process. - */ - foreach(lc, constraints) - { - Constraint *constr = lfirst_node(Constraint, lc); - AttrNumber attnum; - char *conname; - bool is_local = true; - int inhcount = 0; - ListCell *lc2; - - Assert(constr->contype == CONSTR_NOTNULL); - - attnum = get_attnum(RelationGetRelid(rel), - strVal(linitial(constr->keys))); - - /* - * Search in the list of inherited constraints for any entries on the - * same column. - */ - foreach(lc2, old_notnulls) - { - CookedConstraint *old = (CookedConstraint *) lfirst(lc2); - - if (old->attnum == attnum) - { - /* - * If we get a constraint from the parent, having a local NO - * INHERIT one doesn't work. - */ - if (constr->is_no_inherit) - ereport(ERROR, - (errcode(ERRCODE_DATATYPE_MISMATCH), - errmsg("cannot define not-null constraint on column \"%s\" with NO INHERIT", - strVal(linitial(constr->keys))), - errdetail("The column has an inherited not-null constraint."))); - - inhcount++; - old_notnulls = foreach_delete_current(old_notnulls, lc2); - } - } - - /* - * Determine a constraint name, which may have been specified by the - * user, or raise an error if a conflict exists with another - * user-specified name. - */ - if (constr->conname) - { - foreach(lc2, givennames) - { - if (strcmp(lfirst(lc2), constr->conname) == 0) - ereport(ERROR, - errcode(ERRCODE_DUPLICATE_OBJECT), - errmsg("constraint \"%s\" for relation \"%s\" already exists", - constr->conname, - RelationGetRelationName(rel))); - } - - conname = constr->conname; - givennames = lappend(givennames, conname); - } - else - conname = ChooseConstraintName(RelationGetRelationName(rel), - get_attname(RelationGetRelid(rel), - attnum, false), - "not_null", - RelationGetNamespace(rel), - nnnames); - nnnames = lappend(nnnames, conname); - - StoreRelNotNull(rel, conname, - attnum, true, is_local, - inhcount, constr->is_no_inherit); - - nncols = lappend_int(nncols, attnum); - } - - /* - * If any column remains in the old_notnulls list, we must create a not- - * null constraint marked not-local. Because multiple parents could - * specify a not-null constraint for the same column, we must count how - * many there are and add to the original inhcount accordingly, deleting - * elements we've already processed. We sort the list to make it easy. - * - * We don't use foreach() here because we have two nested loops over the - * constraint list, with possible element deletions in the inner one. If - * we used foreach_delete_current() it could only fix up the state of one - * of the loops, so it seems cleaner to use looping over list indexes for - * both loops. Note that any deletion will happen beyond where the outer - * loop is, so its index never needs adjustment. - */ - list_sort(old_notnulls, list_cookedconstr_attnum_cmp); - for (int outerpos = 0; outerpos < list_length(old_notnulls); outerpos++) - { - CookedConstraint *cooked; - char *conname = NULL; - int add_inhcount = 0; - ListCell *lc2; - - cooked = (CookedConstraint *) list_nth(old_notnulls, outerpos); - Assert(cooked->contype == CONSTR_NOTNULL); - - /* - * Preserve the first non-conflicting constraint name we come across, - * if any - */ - if (conname == NULL && cooked->name) - conname = cooked->name; - - for (int restpos = outerpos + 1; restpos < list_length(old_notnulls);) - { - CookedConstraint *other; - - other = (CookedConstraint *) list_nth(old_notnulls, restpos); - if (other->attnum == cooked->attnum) - { - if (conname == NULL && other->name) - conname = other->name; - - add_inhcount++; - old_notnulls = list_delete_nth_cell(old_notnulls, restpos); - } - else - restpos++; - } - - /* If we got a name, make sure it isn't one we've already used */ - if (conname != NULL) - { - foreach(lc2, nnnames) - { - if (strcmp(lfirst(lc2), conname) == 0) - { - conname = NULL; - break; - } - } - } - - /* and choose a name, if needed */ - if (conname == NULL) - conname = ChooseConstraintName(RelationGetRelationName(rel), - get_attname(RelationGetRelid(rel), - cooked->attnum, false), - "not_null", - RelationGetNamespace(rel), - nnnames); - nnnames = lappend(nnnames, conname); - - StoreRelNotNull(rel, conname, cooked->attnum, true, - cooked->is_local, cooked->inhcount + add_inhcount, - cooked->is_no_inherit); - - nncols = lappend_int(nncols, cooked->attnum); - } - - return nncols; -} - /* * Update the count of constraints in the relation's pg_class tuple. * diff --git a/src/backend/catalog/information_schema.sql b/src/backend/catalog/information_schema.sql index 76c78c0d184..c4145131ce4 100644 --- a/src/backend/catalog/information_schema.sql +++ b/src/backend/catalog/information_schema.sql @@ -440,8 +440,9 @@ CREATE VIEW check_constraints AS WHERE pg_has_role(coalesce(c.relowner, t.typowner), 'USAGE') AND con.contype = 'c' - UNION ALL - -- not-null constraints + UNION + -- not-null constraints on domains + SELECT current_database()::information_schema.sql_identifier AS constraint_catalog, rs.nspname::information_schema.sql_identifier AS constraint_schema, con.conname::information_schema.sql_identifier AS constraint_name, @@ -452,7 +453,24 @@ CREATE VIEW check_constraints AS LEFT JOIN pg_type t ON t.oid = con.contypid LEFT JOIN pg_attribute at ON (con.conrelid = at.attrelid AND con.conkey[1] = at.attnum) WHERE pg_has_role(coalesce(c.relowner, t.typowner), 'USAGE'::text) - AND con.contype = 'n'; + AND con.contype = 'n' + + UNION + -- not-null constraints on relations + + SELECT CAST(current_database() AS sql_identifier) AS constraint_catalog, + CAST(n.nspname AS sql_identifier) AS constraint_schema, + CAST(CAST(n.oid AS text) || '_' || CAST(r.oid AS text) || '_' || CAST(a.attnum AS text) || '_not_null' AS sql_identifier) AS constraint_name, -- XXX + CAST(a.attname || ' IS NOT NULL' AS character_data) + AS check_clause + FROM pg_namespace n, pg_class r, pg_attribute a + WHERE n.oid = r.relnamespace + AND r.oid = a.attrelid + AND a.attnum > 0 + AND NOT a.attisdropped + AND a.attnotnull + AND r.relkind IN ('r', 'p') + AND pg_has_role(r.relowner, 'USAGE'); GRANT SELECT ON check_constraints TO PUBLIC; @@ -821,20 +839,6 @@ CREATE VIEW constraint_column_usage AS UNION ALL - /* not-null constraints */ - SELECT DISTINCT nr.nspname, r.relname, r.relowner, a.attname, nc.nspname, c.conname - FROM pg_namespace nr, pg_class r, pg_attribute a, pg_namespace nc, pg_constraint c - WHERE nr.oid = r.relnamespace - AND r.oid = a.attrelid - AND r.oid = c.conrelid - AND a.attnum = c.conkey[1] - AND c.connamespace = nc.oid - AND c.contype = 'n' - AND r.relkind in ('r', 'p') - AND not a.attisdropped - - UNION ALL - /* unique/primary key/foreign key constraints */ SELECT nr.nspname, r.relname, r.relowner, a.attname, nc.nspname, c.conname FROM pg_namespace nr, pg_class r, pg_attribute a, pg_namespace nc, @@ -1835,7 +1839,6 @@ CREATE VIEW table_constraints AS CAST(r.relname AS sql_identifier) AS table_name, CAST( CASE c.contype WHEN 'c' THEN 'CHECK' - WHEN 'n' THEN 'CHECK' WHEN 'f' THEN 'FOREIGN KEY' WHEN 'p' THEN 'PRIMARY KEY' WHEN 'u' THEN 'UNIQUE' END @@ -1863,6 +1866,38 @@ CREATE VIEW table_constraints AS AND (pg_has_role(r.relowner, 'USAGE') -- SELECT privilege omitted, per SQL standard OR has_table_privilege(r.oid, 'INSERT, UPDATE, DELETE, TRUNCATE, REFERENCES, TRIGGER') + OR has_any_column_privilege(r.oid, 'INSERT, UPDATE, REFERENCES') ) + + UNION ALL + + -- not-null constraints + + SELECT CAST(current_database() AS sql_identifier) AS constraint_catalog, + CAST(nr.nspname AS sql_identifier) AS constraint_schema, + CAST(CAST(nr.oid AS text) || '_' || CAST(r.oid AS text) || '_' || CAST(a.attnum AS text) || '_not_null' AS sql_identifier) AS constraint_name, -- XXX + CAST(current_database() AS sql_identifier) AS table_catalog, + CAST(nr.nspname AS sql_identifier) AS table_schema, + CAST(r.relname AS sql_identifier) AS table_name, + CAST('CHECK' AS character_data) AS constraint_type, + CAST('NO' AS yes_or_no) AS is_deferrable, + CAST('NO' AS yes_or_no) AS initially_deferred, + CAST('YES' AS yes_or_no) AS enforced, + CAST(NULL AS yes_or_no) AS nulls_distinct + + FROM pg_namespace nr, + pg_class r, + pg_attribute a + + WHERE nr.oid = r.relnamespace + AND r.oid = a.attrelid + AND a.attnotnull + AND a.attnum > 0 + AND NOT a.attisdropped + AND r.relkind IN ('r', 'p') + AND (NOT pg_is_other_temp_schema(nr.oid)) + AND (pg_has_role(r.relowner, 'USAGE') + -- SELECT privilege omitted, per SQL standard + OR has_table_privilege(r.oid, 'INSERT, UPDATE, DELETE, TRUNCATE, REFERENCES, TRIGGER') OR has_any_column_privilege(r.oid, 'INSERT, UPDATE, REFERENCES') ); GRANT SELECT ON table_constraints TO PUBLIC; diff --git a/src/backend/catalog/pg_constraint.c b/src/backend/catalog/pg_constraint.c index 12a73d5a309..b10e458b449 100644 --- a/src/backend/catalog/pg_constraint.c +++ b/src/backend/catalog/pg_constraint.c @@ -19,10 +19,8 @@ #include "access/htup_details.h" #include "access/sysattr.h" #include "access/table.h" -#include "access/xact.h" #include "catalog/catalog.h" #include "catalog/dependency.h" -#include "catalog/heap.h" #include "catalog/indexing.h" #include "catalog/objectaccess.h" #include "catalog/pg_constraint.h" @@ -566,72 +564,6 @@ ChooseConstraintName(const char *name1, const char *name2, } /* - * Find and return a copy of the pg_constraint tuple that implements a - * validated not-null constraint for the given column of the given relation. - * - * XXX This would be easier if we had pg_attribute.notnullconstr with the OID - * of the constraint that implements the not-null constraint for that column. - * I'm not sure it's worth the catalog bloat and de-normalization, however. - */ -HeapTuple -findNotNullConstraintAttnum(Oid relid, AttrNumber attnum) -{ - Relation pg_constraint; - HeapTuple conTup, - retval = NULL; - SysScanDesc scan; - ScanKeyData key; - - pg_constraint = table_open(ConstraintRelationId, AccessShareLock); - ScanKeyInit(&key, - Anum_pg_constraint_conrelid, - BTEqualStrategyNumber, F_OIDEQ, - ObjectIdGetDatum(relid)); - scan = systable_beginscan(pg_constraint, ConstraintRelidTypidNameIndexId, - true, NULL, 1, &key); - - while (HeapTupleIsValid(conTup = systable_getnext(scan))) - { - Form_pg_constraint con = (Form_pg_constraint) GETSTRUCT(conTup); - AttrNumber conkey; - - /* - * We're looking for a NOTNULL constraint that's marked validated, - * with the column we're looking for as the sole element in conkey. - */ - if (con->contype != CONSTRAINT_NOTNULL) - continue; - if (!con->convalidated) - continue; - - conkey = extractNotNullColumn(conTup); - if (conkey != attnum) - continue; - - /* Found it */ - retval = heap_copytuple(conTup); - break; - } - - systable_endscan(scan); - table_close(pg_constraint, AccessShareLock); - - return retval; -} - -/* - * Find and return the pg_constraint tuple that implements a validated - * not-null constraint for the given column of the given relation. - */ -HeapTuple -findNotNullConstraint(Oid relid, const char *colname) -{ - AttrNumber attnum = get_attnum(relid, colname); - - return findNotNullConstraintAttnum(relid, attnum); -} - -/* * Find and return the pg_constraint tuple that implements a validated * not-null constraint for the given domain. */ @@ -676,263 +608,6 @@ findDomainNotNullConstraint(Oid typid) } /* - * Given a pg_constraint tuple for a not-null constraint, return the column - * number it is for. - */ -AttrNumber -extractNotNullColumn(HeapTuple constrTup) -{ - AttrNumber colnum; - Datum adatum; - ArrayType *arr; - - /* only tuples for not-null constraints should be given */ - Assert(((Form_pg_constraint) GETSTRUCT(constrTup))->contype == CONSTRAINT_NOTNULL); - - adatum = SysCacheGetAttrNotNull(CONSTROID, constrTup, - Anum_pg_constraint_conkey); - arr = DatumGetArrayTypeP(adatum); /* ensure not toasted */ - if (ARR_NDIM(arr) != 1 || - ARR_HASNULL(arr) || - ARR_ELEMTYPE(arr) != INT2OID || - ARR_DIMS(arr)[0] != 1) - elog(ERROR, "conkey is not a 1-D smallint array"); - - memcpy(&colnum, ARR_DATA_PTR(arr), sizeof(AttrNumber)); - - if ((Pointer) arr != DatumGetPointer(adatum)) - pfree(arr); /* free de-toasted copy, if any */ - - return colnum; -} - -/* - * AdjustNotNullInheritance1 - * Adjust inheritance count for a single not-null constraint - * - * If no not-null constraint is found for the column, return 0. - * Caller can create one. - * If the constraint does exist and it's inheritable, adjust its - * inheritance count (and possibly islocal status) and return 1. - * No further action needs to be taken. - * If the constraint exists but is marked NO INHERIT, adjust it as above - * and reset connoinherit to false, and return -1. Caller is - * responsible for adding the same constraint to the children, if any. - */ -int -AdjustNotNullInheritance1(Oid relid, AttrNumber attnum, int count, - bool is_no_inherit, bool allow_noinherit_change) -{ - HeapTuple tup; - - Assert(count >= 0); - - tup = findNotNullConstraintAttnum(relid, attnum); - if (HeapTupleIsValid(tup)) - { - Relation pg_constraint; - Form_pg_constraint conform; - int retval = 1; - - pg_constraint = table_open(ConstraintRelationId, RowExclusiveLock); - conform = (Form_pg_constraint) GETSTRUCT(tup); - - /* - * If we're asked for a NO INHERIT constraint and this relation - * already has an inheritable one, throw an error. - */ - if (is_no_inherit && !conform->connoinherit) - ereport(ERROR, - errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), - errmsg("cannot change NO INHERIT status of NOT NULL constraint \"%s\" on relation \"%s\"", - NameStr(conform->conname), get_rel_name(relid))); - - /* - * If the constraint already exists in this relation but it's marked - * NO INHERIT, we can just remove that flag (provided caller allows - * such a change), and instruct caller to recurse to add the - * constraint to children. - */ - if (!is_no_inherit && conform->connoinherit) - { - if (!allow_noinherit_change) - ereport(ERROR, - errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), - errmsg("cannot change NO INHERIT status of NOT NULL constraint \"%s\" on relation \"%s\"", - NameStr(conform->conname), get_rel_name(relid))); - - conform->connoinherit = false; - retval = -1; /* caller must add constraint on child rels */ - } - - if (count > 0) - conform->coninhcount += count; - - /* sanity check */ - if (conform->coninhcount < 0) - elog(ERROR, "invalid inhcount %d for constraint \"%s\" on relation \"%s\"", - conform->coninhcount, NameStr(conform->conname), - get_rel_name(relid)); - - /* - * If the constraint is no longer inherited, mark it local. It's - * arguable that we should drop it instead, but it's hard to see that - * being better. The user can drop it manually later. - */ - if (conform->coninhcount == 0) - conform->conislocal = true; - - CatalogTupleUpdate(pg_constraint, &tup->t_self, tup); - - table_close(pg_constraint, RowExclusiveLock); - - return retval; - } - - return 0; -} - -/* - * AdjustNotNullInheritance - * Adjust not-null constraints' inhcount/islocal for - * ALTER TABLE [NO] INHERITS - * - * Mark the NOT NULL constraints for the given relation columns as - * inherited, so that they can't be dropped. - * - * Caller must have checked beforehand that attnotnull was set for all - * columns. However, some of those could be set because of a primary - * key, so throw a proper user-visible error if one is not found. - */ -void -AdjustNotNullInheritance(Oid relid, Bitmapset *columns, int count) -{ - Relation pg_constraint; - int attnum; - - pg_constraint = table_open(ConstraintRelationId, RowExclusiveLock); - - /* - * Scan the set of columns and bump inhcount for each. - */ - attnum = -1; - while ((attnum = bms_next_member(columns, attnum)) >= 0) - { - HeapTuple tup; - Form_pg_constraint conform; - - tup = findNotNullConstraintAttnum(relid, attnum); - if (!HeapTupleIsValid(tup)) - ereport(ERROR, - errcode(ERRCODE_DATATYPE_MISMATCH), - errmsg("column \"%s\" in child table must be marked NOT NULL", - get_attname(relid, attnum, - false))); - - conform = (Form_pg_constraint) GETSTRUCT(tup); - conform->coninhcount += count; - if (conform->coninhcount < 0) - elog(ERROR, "invalid inhcount %d for constraint \"%s\" on relation \"%s\"", - conform->coninhcount, NameStr(conform->conname), - get_rel_name(relid)); - - /* - * If the constraints are no longer inherited, mark them local. It's - * arguable that we should drop them instead, but it's hard to see - * that being better. The user can drop it manually later. - */ - if (conform->coninhcount == 0) - conform->conislocal = true; - - CatalogTupleUpdate(pg_constraint, &tup->t_self, tup); - } - - table_close(pg_constraint, RowExclusiveLock); -} - -/* - * RelationGetNotNullConstraints - * Return the list of not-null constraints for the given rel - * - * Caller can request cooked constraints, or raw. - * - * This is seldom needed, so we just scan pg_constraint each time. - * - * XXX This is only used to create derived tables, so NO INHERIT constraints - * are always skipped. - */ -List * -RelationGetNotNullConstraints(Oid relid, bool cooked) -{ - List *notnulls = NIL; - Relation constrRel; - HeapTuple htup; - SysScanDesc conscan; - ScanKeyData skey; - - constrRel = table_open(ConstraintRelationId, AccessShareLock); - ScanKeyInit(&skey, - Anum_pg_constraint_conrelid, - BTEqualStrategyNumber, F_OIDEQ, - ObjectIdGetDatum(relid)); - conscan = systable_beginscan(constrRel, ConstraintRelidTypidNameIndexId, true, - NULL, 1, &skey); - - while (HeapTupleIsValid(htup = systable_getnext(conscan))) - { - Form_pg_constraint conForm = (Form_pg_constraint) GETSTRUCT(htup); - AttrNumber colnum; - - if (conForm->contype != CONSTRAINT_NOTNULL) - continue; - if (conForm->connoinherit) - continue; - - colnum = extractNotNullColumn(htup); - - if (cooked) - { - CookedConstraint *cooked; - - cooked = (CookedConstraint *) palloc(sizeof(CookedConstraint)); - - cooked->contype = CONSTR_NOTNULL; - cooked->name = pstrdup(NameStr(conForm->conname)); - cooked->attnum = colnum; - cooked->expr = NULL; - cooked->skip_validation = false; - cooked->is_local = true; - cooked->inhcount = 0; - cooked->is_no_inherit = conForm->connoinherit; - - notnulls = lappend(notnulls, cooked); - } - else - { - Constraint *constr; - - constr = makeNode(Constraint); - constr->contype = CONSTR_NOTNULL; - constr->conname = pstrdup(NameStr(conForm->conname)); - constr->deferrable = false; - constr->initdeferred = false; - constr->location = -1; - constr->keys = list_make1(makeString(get_attname(relid, colnum, - false))); - constr->skip_validation = false; - constr->initially_valid = true; - notnulls = lappend(notnulls, constr); - } - } - - systable_endscan(conscan); - table_close(constrRel, AccessShareLock); - - return notnulls; -} - - -/* * Delete a single constraint record. */ void @@ -941,8 +616,6 @@ RemoveConstraintById(Oid conId) Relation conDesc; HeapTuple tup; Form_pg_constraint con; - bool dropping_pk = false; - List *unconstrained_cols = NIL; conDesc = table_open(ConstraintRelationId, RowExclusiveLock); @@ -967,9 +640,7 @@ RemoveConstraintById(Oid conId) /* * We need to update the relchecks count if it is a check constraint * being dropped. This update will force backends to rebuild relcache - * entries when we commit. For not-null and primary key constraints, - * obtain the list of columns affected, so that we can reset their - * attnotnull flags below. + * entries when we commit. */ if (con->contype == CONSTRAINT_CHECK) { @@ -996,36 +667,6 @@ RemoveConstraintById(Oid conId) table_close(pgrel, RowExclusiveLock); } - else if (con->contype == CONSTRAINT_NOTNULL) - { - unconstrained_cols = list_make1_int(extractNotNullColumn(tup)); - } - else if (con->contype == CONSTRAINT_PRIMARY) - { - Datum adatum; - ArrayType *arr; - int numkeys; - bool isNull; - int16 *attnums; - - dropping_pk = true; - - adatum = heap_getattr(tup, Anum_pg_constraint_conkey, - RelationGetDescr(conDesc), &isNull); - if (isNull) - elog(ERROR, "null conkey for constraint %u", con->oid); - arr = DatumGetArrayTypeP(adatum); /* ensure not toasted */ - numkeys = ARR_DIMS(arr)[0]; - if (ARR_NDIM(arr) != 1 || - numkeys < 0 || - ARR_HASNULL(arr) || - ARR_ELEMTYPE(arr) != INT2OID) - elog(ERROR, "conkey is not a 1-D smallint array"); - attnums = (int16 *) ARR_DATA_PTR(arr); - - for (int i = 0; i < numkeys; i++) - unconstrained_cols = lappend_int(unconstrained_cols, attnums[i]); - } /* Keep lock on constraint's rel until end of xact */ table_close(rel, NoLock); @@ -1045,86 +686,6 @@ RemoveConstraintById(Oid conId) /* Fry the constraint itself */ CatalogTupleDelete(conDesc, &tup->t_self); - /* - * If this was a NOT NULL or the primary key, the constrained columns must - * have had pg_attribute.attnotnull set. See if we need to reset it, and - * do so. - */ - if (unconstrained_cols != NIL) - { - Relation tablerel; - Relation attrel; - Bitmapset *pkcols; - ListCell *lc; - - /* Make the above deletion visible */ - CommandCounterIncrement(); - - tablerel = table_open(con->conrelid, NoLock); /* already have lock */ - attrel = table_open(AttributeRelationId, RowExclusiveLock); - - /* - * We want to test columns for their presence in the primary key, but - * only if we're not dropping it. - */ - pkcols = dropping_pk ? NULL : - RelationGetIndexAttrBitmap(tablerel, - INDEX_ATTR_BITMAP_PRIMARY_KEY); - - foreach(lc, unconstrained_cols) - { - AttrNumber attnum = lfirst_int(lc); - HeapTuple atttup; - HeapTuple contup; - Bitmapset *ircols; - Form_pg_attribute attForm; - - /* - * Obtain pg_attribute tuple and verify conditions on it. We use - * a copy we can scribble on. - */ - atttup = SearchSysCacheCopyAttNum(con->conrelid, attnum); - if (!HeapTupleIsValid(atttup)) - elog(ERROR, "cache lookup failed for attribute %d of relation %u", - attnum, con->conrelid); - attForm = (Form_pg_attribute) GETSTRUCT(atttup); - - /* - * Since the above deletion has been made visible, we can now - * search for any remaining constraints setting this column as - * not-nullable; if we find any, no need to reset attnotnull. - */ - if (bms_is_member(attnum - FirstLowInvalidHeapAttributeNumber, - pkcols)) - continue; - contup = findNotNullConstraintAttnum(con->conrelid, attnum); - if (contup) - continue; - - /* - * Also no reset if the column is in the replica identity or it's - * a generated column - */ - if (attForm->attidentity != '\0') - continue; - ircols = RelationGetIndexAttrBitmap(tablerel, - INDEX_ATTR_BITMAP_IDENTITY_KEY); - if (bms_is_member(attnum - FirstLowInvalidHeapAttributeNumber, - ircols)) - continue; - - /* Reset attnotnull */ - if (attForm->attnotnull) - { - attForm->attnotnull = false; - CatalogTupleUpdate(attrel, &atttup->t_self, atttup); - } - } - - table_close(attrel, RowExclusiveLock); - table_close(tablerel, NoLock); - } - /* Clean up */ ReleaseSysCache(tup); table_close(conDesc, RowExclusiveLock); diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c index de0d911b468..79c9c031833 100644 --- a/src/backend/commands/tablecmds.c +++ b/src/backend/commands/tablecmds.c @@ -151,7 +151,6 @@ typedef enum AlterTablePass AT_PASS_ALTER_TYPE, /* ALTER COLUMN TYPE */ AT_PASS_ADD_COL, /* ADD COLUMN */ AT_PASS_SET_EXPRESSION, /* ALTER SET EXPRESSION */ - AT_PASS_OLD_COL_ATTRS, /* re-install attnotnull */ AT_PASS_OLD_INDEX, /* re-add existing indexes */ AT_PASS_OLD_CONSTR, /* re-add existing constraints */ /* We could support a RENAME COLUMN pass here, but not currently used */ @@ -362,8 +361,7 @@ static void truncate_check_activity(Relation rel); static void RangeVarCallbackForTruncate(const RangeVar *relation, Oid relId, Oid oldRelId, void *arg); static List *MergeAttributes(List *columns, const List *supers, char relpersistence, - bool is_partition, List **supconstr, - List **supnotnulls); + bool is_partition, List **supconstr); static List *MergeCheckConstraint(List *constraints, const char *name, Node *expr); static void MergeChildAttribute(List *inh_columns, int exist_attno, int newcol_attno, const ColumnDef *newdef); static ColumnDef *MergeInheritedAttribute(List *inh_columns, int exist_attno, const ColumnDef *newdef); @@ -447,16 +445,16 @@ static bool check_for_column_name_collision(Relation rel, const char *colname, bool if_not_exists); static void add_column_datatype_dependency(Oid relid, int32 attnum, Oid typid); static void add_column_collation_dependency(Oid relid, int32 attnum, Oid collid); -static ObjectAddress ATExecDropNotNull(Relation rel, const char *colName, bool recurse, - LOCKMODE lockmode); -static bool set_attnotnull(List **wqueue, Relation rel, - AttrNumber attnum, bool recurse, LOCKMODE lockmode); -static ObjectAddress ATExecSetNotNull(List **wqueue, Relation rel, - char *constrname, char *colName, - bool recurse, bool recursing, - List **readyRels, LOCKMODE lockmode); -static ObjectAddress ATExecSetAttNotNull(List **wqueue, Relation rel, - const char *colName, LOCKMODE lockmode); +static void ATPrepDropNotNull(Relation rel, bool recurse, bool recursing); +static ObjectAddress ATExecDropNotNull(Relation rel, const char *colName, LOCKMODE lockmode); +static void ATPrepSetNotNull(List **wqueue, Relation rel, + AlterTableCmd *cmd, bool recurse, bool recursing, + LOCKMODE lockmode, + AlterTableUtilityContext *context); +static ObjectAddress ATExecSetNotNull(AlteredTableInfo *tab, Relation rel, + const char *colName, LOCKMODE lockmode); +static void ATExecCheckNotNull(AlteredTableInfo *tab, Relation rel, + const char *colName, LOCKMODE lockmode); static bool NotNullImpliedByRelConstraints(Relation rel, Form_pg_attribute attr); static bool ConstraintImpliedByRelConstraint(Relation scanrel, List *testConstraint, List *provenConstraint); @@ -488,8 +486,6 @@ static ObjectAddress ATExecDropColumn(List **wqueue, Relation rel, const char *c bool recurse, bool recursing, bool missing_ok, LOCKMODE lockmode, ObjectAddresses *addrs); -static void ATPrepAddPrimaryKey(List **wqueue, Relation rel, AlterTableCmd *cmd, - LOCKMODE lockmode, AlterTableUtilityContext *context); static ObjectAddress ATExecAddIndex(AlteredTableInfo *tab, Relation rel, IndexStmt *stmt, bool is_rebuild, LOCKMODE lockmode); static ObjectAddress ATExecAddStatistics(AlteredTableInfo *tab, Relation rel, @@ -501,11 +497,11 @@ static ObjectAddress ATExecAddConstraint(List **wqueue, static char *ChooseForeignKeyConstraintNameAddition(List *colnames); static ObjectAddress ATExecAddIndexConstraint(AlteredTableInfo *tab, Relation rel, IndexStmt *stmt, LOCKMODE lockmode); -static ObjectAddress ATAddCheckNNConstraint(List **wqueue, - AlteredTableInfo *tab, Relation rel, - Constraint *constr, - bool recurse, bool recursing, bool is_readd, - LOCKMODE lockmode); +static ObjectAddress ATAddCheckConstraint(List **wqueue, + AlteredTableInfo *tab, Relation rel, + Constraint *constr, + bool recurse, bool recursing, bool is_readd, + LOCKMODE lockmode); static ObjectAddress ATAddForeignKeyConstraint(List **wqueue, AlteredTableInfo *tab, Relation rel, Constraint *fkconstraint, bool recurse, bool recursing, @@ -562,13 +558,9 @@ static void GetForeignKeyCheckTriggers(Relation trigrel, Oid *insertTriggerOid, Oid *updateTriggerOid); static void ATExecDropConstraint(Relation rel, const char *constrName, - DropBehavior behavior, bool recurse, + DropBehavior behavior, + bool recurse, bool recursing, bool missing_ok, LOCKMODE lockmode); -static ObjectAddress dropconstraint_internal(Relation rel, - HeapTuple constraintTup, DropBehavior behavior, - bool recurse, bool recursing, - bool missing_ok, List **readyRels, - LOCKMODE lockmode); static void ATPrepAlterColumnType(List **wqueue, AlteredTableInfo *tab, Relation rel, bool recurse, bool recursing, @@ -644,8 +636,6 @@ static void ComputePartitionAttrs(ParseState *pstate, Relation rel, List *partPa static void CreateInheritance(Relation child_rel, Relation parent_rel, bool ispartition); static void RemoveInheritance(Relation child_rel, Relation parent_rel, bool expect_detached); -static void ATInheritAdjustNotNulls(Relation parent_rel, Relation child_rel, - int inhcount); static ObjectAddress ATExecAttachPartition(List **wqueue, Relation rel, PartitionCmd *cmd, AlterTableUtilityContext *context); @@ -667,7 +657,6 @@ static ObjectAddress ATExecAttachPartitionIdx(List **wqueue, Relation parentIdx, static void validatePartitionedIndex(Relation partedIdx, Relation partedTbl); static void refuseDupeIndexAttach(Relation parentIdx, Relation partIdx, Relation partitionTbl); -static void verifyPartitionIndexNotNull(IndexInfo *iinfo, Relation partIdx); static List *GetParentedForeignKeyRefs(Relation partition); static void ATDetachCheckNoForeignKeyRefs(Relation partition); static char GetAttributeCompression(Oid atttypid, const char *compression); @@ -710,10 +699,8 @@ DefineRelation(CreateStmt *stmt, char relkind, Oid ownerId, TupleDesc descriptor; List *inheritOids; List *old_constraints; - List *old_notnulls; List *rawDefaults; List *cookedDefaults; - List *nncols; Datum reloptions; ListCell *listptr; AttrNumber attnum; @@ -898,13 +885,12 @@ DefineRelation(CreateStmt *stmt, char relkind, Oid ownerId, MergeAttributes(stmt->tableElts, inheritOids, stmt->relation->relpersistence, stmt->partbound != NULL, - &old_constraints, &old_notnulls); + &old_constraints); /* * Create a tuple descriptor from the relation schema. Note that this - * deals with column names, types, and in-descriptor NOT NULL flags, but - * not default values, NOT NULL or CHECK constraints; we handle those - * below. + * deals with column names, types, and not-null constraints, but not + * default values or CHECK constraints; we handle those below. */ descriptor = BuildDescForRelation(stmt->tableElts); @@ -1276,17 +1262,6 @@ DefineRelation(CreateStmt *stmt, char relkind, Oid ownerId, AddRelationNewConstraints(rel, NIL, stmt->constraints, true, true, false, queryString); - /* - * Finally, merge the not-null constraints that are declared directly with - * those that come from parent relations (making sure to count inheritance - * appropriately for each), create them, and set the attnotnull flag on - * columns that don't yet have it. - */ - nncols = AddRelationNotNullConstraints(rel, stmt->nnconstraints, - old_notnulls); - foreach(listptr, nncols) - set_attnotnull(NULL, rel, lfirst_int(listptr), false, NoLock); - ObjectAddressSet(address, RelationRelationId, relationId); /* @@ -2437,8 +2412,6 @@ storage_name(char c) * Output arguments: * 'supconstr' receives a list of constraints belonging to the parents, * updated as necessary to be valid for the child. - * 'supnotnulls' receives a list of CookedConstraints that corresponds to - * constraints coming from inheritance parents. * * Return value: * Completed schema list. @@ -2469,10 +2442,7 @@ storage_name(char c) * * Constraints (including not-null constraints) for the child table * are the union of all relevant constraints, from both the child schema - * and parent tables. In addition, in legacy inheritance, each column that - * appears in a primary key in any of the parents also gets a NOT NULL - * constraint (partitioning doesn't need this, because the PK itself gets - * inherited.) + * and parent tables. * * The default value for a child column is defined as: * (1) If the child schema specifies a default, that value is used. @@ -2491,11 +2461,10 @@ storage_name(char c) */ static List * MergeAttributes(List *columns, const List *supers, char relpersistence, - bool is_partition, List **supconstr, List **supnotnulls) + bool is_partition, List **supconstr) { List *inh_columns = NIL; List *constraints = NIL; - List *nnconstraints = NIL; bool have_bogus_defaults = false; int child_attno; static Node bogus_marker = {0}; /* marks conflicting defaults */ @@ -2606,11 +2575,8 @@ MergeAttributes(List *columns, const List *supers, char relpersistence, AttrMap *newattmap; List *inherited_defaults; List *cols_with_defaults; - List *nnconstrs; ListCell *lc1; ListCell *lc2; - Bitmapset *pkattrs; - Bitmapset *nncols = NULL; /* caller already got lock */ relation = table_open(parent, NoLock); @@ -2698,20 +2664,6 @@ MergeAttributes(List *columns, const List *supers, char relpersistence, /* We can't process inherited defaults until newattmap is complete. */ inherited_defaults = cols_with_defaults = NIL; - /* - * All columns that are part of the parent's primary key need to be - * NOT NULL; if partition just the attnotnull bit, otherwise a full - * constraint (if they don't have one already). Also, we request - * attnotnull on columns that have a not-null constraint that's not - * marked NO INHERIT. - */ - pkattrs = RelationGetIndexAttrBitmap(relation, - INDEX_ATTR_BITMAP_PRIMARY_KEY); - nnconstrs = RelationGetNotNullConstraints(RelationGetRelid(relation), true); - foreach(lc1, nnconstrs) - nncols = bms_add_member(nncols, - ((CookedConstraint *) lfirst(lc1))->attnum); - for (AttrNumber parent_attno = 1; parent_attno <= tupleDesc->natts; parent_attno++) { @@ -2733,6 +2685,7 @@ MergeAttributes(List *columns, const List *supers, char relpersistence, */ newdef = makeColumnDef(attributeName, attribute->atttypid, attribute->atttypmod, attribute->attcollation); + newdef->is_not_null = attribute->attnotnull; newdef->storage = attribute->attstorage; newdef->generated = attribute->attgenerated; if (CompressionMethodIsValid(attribute->attcompression)) @@ -2777,45 +2730,10 @@ MergeAttributes(List *columns, const List *supers, char relpersistence, inh_columns = lappend(inh_columns, newdef); newattmap->attnums[parent_attno - 1] = ++child_attno; - mergeddef = newdef; } /* - * mark attnotnull if parent has it and it's not NO INHERIT - */ - if (bms_is_member(parent_attno, nncols) || - bms_is_member(parent_attno - FirstLowInvalidHeapAttributeNumber, - pkattrs)) - mergeddef->is_not_null = true; - - /* - * In regular inheritance, columns in the parent's primary key get - * an extra not-null constraint. Partitioning doesn't need this, - * because the PK itself is going to be cloned to the partition. - */ - if (!is_partition && - bms_is_member(parent_attno - - FirstLowInvalidHeapAttributeNumber, - pkattrs)) - { - CookedConstraint *nn; - - nn = palloc(sizeof(CookedConstraint)); - nn->contype = CONSTR_NOTNULL; - nn->conoid = InvalidOid; - nn->name = NULL; - nn->attnum = newattmap->attnums[parent_attno - 1]; - nn->expr = NULL; - nn->skip_validation = false; - nn->is_local = false; - nn->inhcount = 1; - nn->is_no_inherit = false; - - nnconstraints = lappend(nnconstraints, nn); - } - - /* * Locate default/generation expression if any */ if (attribute->atthasdef) @@ -2926,23 +2844,6 @@ MergeAttributes(List *columns, const List *supers, char relpersistence, } } - /* - * Also copy the not-null constraints from this parent. The - * attnotnull markings were already installed above. - */ - foreach(lc1, nnconstrs) - { - CookedConstraint *nn = lfirst(lc1); - - Assert(nn->contype == CONSTR_NOTNULL); - - nn->attnum = newattmap->attnums[nn->attnum - 1]; - nn->is_local = false; - nn->inhcount = 1; - - nnconstraints = lappend(nnconstraints, nn); - } - free_attrmap(newattmap); /* @@ -3013,7 +2914,8 @@ MergeAttributes(List *columns, const List *supers, char relpersistence, /* * Now that we have the column definition list for a partition, we can * check whether the columns referenced in the column constraint specs - * actually exist. Also, merge column defaults. + * actually exist. Also, we merge parent's not-null constraints and + * defaults into each corresponding column definition. */ if (is_partition) { @@ -3030,6 +2932,7 @@ MergeAttributes(List *columns, const List *supers, char relpersistence, if (strcmp(coldef->colname, restdef->colname) == 0) { found = true; + coldef->is_not_null |= restdef->is_not_null; /* * Check for conflicts related to generated columns. @@ -3118,7 +3021,6 @@ MergeAttributes(List *columns, const List *supers, char relpersistence, } *supconstr = constraints; - *supnotnulls = nnconstraints; return columns; } @@ -3400,6 +3302,11 @@ MergeInheritedAttribute(List *inh_columns, format_type_with_typemod(newtypeid, newtypmod)))); /* + * Merge of not-null constraints = OR 'em together + */ + prevdef->is_not_null |= newdef->is_not_null; + + /* * Must have the same collation */ prevcollid = GetColumnDefCollation(NULL, prevdef, prevtypeid); @@ -4022,10 +3929,7 @@ rename_constraint_internal(Oid myrelid, constraintOid); con = (Form_pg_constraint) GETSTRUCT(tuple); - if (myrelid && - (con->contype == CONSTRAINT_CHECK || - con->contype == CONSTRAINT_NOTNULL) && - !con->connoinherit) + if (myrelid && con->contype == CONSTRAINT_CHECK && !con->connoinherit) { if (recurse) { @@ -4610,7 +4514,6 @@ AlterTableGetLockLevel(List *cmds) case AT_AddIndexConstraint: case AT_ReplicaIdentity: case AT_SetNotNull: - case AT_SetAttNotNull: case AT_EnableRowSecurity: case AT_DisableRowSecurity: case AT_ForceRowSecurity: @@ -4758,6 +4661,15 @@ AlterTableGetLockLevel(List *cmds) cmd_lockmode = AccessExclusiveLock; break; + case AT_CheckNotNull: + + /* + * This only examines the table's schema; but lock must be + * strong enough to prevent concurrent DROP NOT NULL. + */ + cmd_lockmode = AccessShareLock; + break; + default: /* oops */ elog(ERROR, "unrecognized alter table type: %d", (int) cmd->subtype); @@ -4915,23 +4827,21 @@ ATPrepCmd(List **wqueue, Relation rel, AlterTableCmd *cmd, break; case AT_DropNotNull: /* ALTER COLUMN DROP NOT NULL */ ATSimplePermissions(cmd->subtype, rel, ATT_TABLE | ATT_FOREIGN_TABLE); - /* Set up recursion for phase 2; no other prep needed */ - if (recurse) - cmd->recurse = true; + ATPrepDropNotNull(rel, recurse, recursing); + ATSimpleRecursion(wqueue, rel, cmd, recurse, lockmode, context); pass = AT_PASS_DROP; break; case AT_SetNotNull: /* ALTER COLUMN SET NOT NULL */ ATSimplePermissions(cmd->subtype, rel, ATT_TABLE | ATT_FOREIGN_TABLE); - /* Set up recursion for phase 2; no other prep needed */ - if (recurse) - cmd->recurse = true; + /* Need command-specific recursion decision */ + ATPrepSetNotNull(wqueue, rel, cmd, recurse, recursing, + lockmode, context); pass = AT_PASS_COL_ATTRS; break; - case AT_SetAttNotNull: /* set pg_attribute.attnotnull without adding - * a constraint */ + case AT_CheckNotNull: /* check column is already marked NOT NULL */ ATSimplePermissions(cmd->subtype, rel, ATT_TABLE | ATT_FOREIGN_TABLE); - /* Need command-specific recursion decision */ ATSimpleRecursion(wqueue, rel, cmd, recurse, lockmode, context); + /* No command-specific prep needed */ pass = AT_PASS_COL_ATTRS; break; case AT_SetExpression: /* ALTER COLUMN SET EXPRESSION */ @@ -4985,12 +4895,10 @@ ATPrepCmd(List **wqueue, Relation rel, AlterTableCmd *cmd, break; case AT_AddConstraint: /* ADD CONSTRAINT */ ATSimplePermissions(cmd->subtype, rel, ATT_TABLE | ATT_FOREIGN_TABLE); + /* Recursion occurs during execution phase */ + /* No command-specific prep needed except saving recurse flag */ if (recurse) - { - /* recurses at exec time; lock descendants and set flag */ - (void) find_all_inheritors(RelationGetRelid(rel), lockmode, NULL); cmd->recurse = true; - } pass = AT_PASS_ADD_CONSTR; break; case AT_AddIndexConstraint: /* ADD CONSTRAINT USING INDEX */ @@ -5320,14 +5228,13 @@ ATExecCmd(List **wqueue, AlteredTableInfo *tab, address = ATExecDropIdentity(rel, cmd->name, cmd->missing_ok, lockmode, cmd->recurse, false); break; case AT_DropNotNull: /* ALTER COLUMN DROP NOT NULL */ - address = ATExecDropNotNull(rel, cmd->name, cmd->recurse, lockmode); + address = ATExecDropNotNull(rel, cmd->name, lockmode); break; case AT_SetNotNull: /* ALTER COLUMN SET NOT NULL */ - address = ATExecSetNotNull(wqueue, rel, NULL, cmd->name, - cmd->recurse, false, NULL, lockmode); + address = ATExecSetNotNull(tab, rel, cmd->name, lockmode); break; - case AT_SetAttNotNull: /* set pg_attribute.attnotnull */ - address = ATExecSetAttNotNull(wqueue, rel, cmd->name, lockmode); + case AT_CheckNotNull: /* check column is already marked NOT NULL */ + ATExecCheckNotNull(tab, rel, cmd->name, lockmode); break; case AT_SetExpression: address = ATExecSetExpression(tab, rel, cmd->name, cmd->def, lockmode); @@ -5410,7 +5317,7 @@ ATExecCmd(List **wqueue, AlteredTableInfo *tab, break; case AT_DropConstraint: /* DROP CONSTRAINT */ ATExecDropConstraint(rel, cmd->name, cmd->behavior, - cmd->recurse, + cmd->recurse, false, cmd->missing_ok, lockmode); break; case AT_AlterColumnType: /* ALTER COLUMN TYPE */ @@ -5689,23 +5596,21 @@ ATParseTransformCmd(List **wqueue, AlteredTableInfo *tab, Relation rel, */ switch (cmd2->subtype) { - case AT_SetAttNotNull: - ATSimpleRecursion(wqueue, rel, cmd2, recurse, lockmode, context); + case AT_SetNotNull: + /* Need command-specific recursion decision */ + ATPrepSetNotNull(wqueue, rel, cmd2, + recurse, false, + lockmode, context); pass = AT_PASS_COL_ATTRS; break; case AT_AddIndex: - - /* - * A primary key on an inheritance parent needs supporting NOT - * NULL constraint on its children; enqueue commands to create - * those or mark them inherited if they already exist. - */ - ATPrepAddPrimaryKey(wqueue, rel, cmd2, lockmode, context); + /* This command never recurses */ + /* No command-specific prep needed */ pass = AT_PASS_ADD_INDEX; break; case AT_AddIndexConstraint: - /* as above */ - ATPrepAddPrimaryKey(wqueue, rel, cmd2, lockmode, context); + /* This command never recurses */ + /* No command-specific prep needed */ pass = AT_PASS_ADD_INDEXCONSTR; break; case AT_AddConstraint: @@ -6372,7 +6277,6 @@ ATRewriteTable(AlteredTableInfo *tab, Oid OIDNewHeap, LOCKMODE lockmode) RelationGetRelationName(oldrel)), errtableconstraint(oldrel, con->name))); break; - case CONSTR_NOTNULL: case CONSTR_FOREIGN: /* Nothing to do here */ break; @@ -6482,12 +6386,12 @@ alter_table_type_to_string(AlterTableType cmdtype) return "ALTER COLUMN ... DROP NOT NULL"; case AT_SetNotNull: return "ALTER COLUMN ... SET NOT NULL"; - case AT_SetAttNotNull: - return NULL; /* not real grammar */ case AT_SetExpression: return "ALTER COLUMN ... SET EXPRESSION"; case AT_DropExpression: return "ALTER COLUMN ... DROP EXPRESSION"; + case AT_CheckNotNull: + return NULL; /* not real grammar */ case AT_SetStatistics: return "ALTER COLUMN ... SET STATISTICS"; case AT_SetOptions: @@ -7570,21 +7474,41 @@ add_column_collation_dependency(Oid relid, int32 attnum, Oid collid) /* * ALTER TABLE ALTER COLUMN DROP NOT NULL - * + */ + +static void +ATPrepDropNotNull(Relation rel, bool recurse, bool recursing) +{ + /* + * If the parent is a partitioned table, like check constraints, we do not + * support removing the NOT NULL while partitions exist. + */ + if (rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE) + { + PartitionDesc partdesc = RelationGetPartitionDesc(rel, true); + + Assert(partdesc != NULL); + if (partdesc->nparts > 0 && !recurse && !recursing) + ereport(ERROR, + (errcode(ERRCODE_INVALID_TABLE_DEFINITION), + errmsg("cannot remove constraint from only the partitioned table when partitions exist"), + errhint("Do not specify the ONLY keyword."))); + } +} + +/* * Return the address of the modified column. If the column was already * nullable, InvalidObjectAddress is returned. */ static ObjectAddress -ATExecDropNotNull(Relation rel, const char *colName, bool recurse, - LOCKMODE lockmode) +ATExecDropNotNull(Relation rel, const char *colName, LOCKMODE lockmode) { HeapTuple tuple; - HeapTuple conTup; Form_pg_attribute attTup; AttrNumber attnum; Relation attr_rel; + List *indexoidlist; ObjectAddress address; - List *readyRels; /* * lookup the attribute @@ -7599,15 +7523,6 @@ ATExecDropNotNull(Relation rel, const char *colName, bool recurse, colName, RelationGetRelationName(rel)))); attTup = (Form_pg_attribute) GETSTRUCT(tuple); attnum = attTup->attnum; - ObjectAddressSubSet(address, RelationRelationId, - RelationGetRelid(rel), attnum); - - /* If the column is already nullable there's nothing to do. */ - if (!attTup->attnotnull) - { - table_close(attr_rel, RowExclusiveLock); - return InvalidObjectAddress; - } /* Prevent them from altering a system attribute */ if (attnum <= 0) @@ -7623,37 +7538,60 @@ ATExecDropNotNull(Relation rel, const char *colName, bool recurse, colName, RelationGetRelationName(rel)))); /* - * It's not OK to remove a constraint only for the parent and leave it in - * the children, so disallow that. + * Check that the attribute is not in a primary key or in an index used as + * a replica identity. + * + * Note: we'll throw error even if the pkey index is not valid. */ - if (!recurse) + + /* Loop over all indexes on the relation */ + indexoidlist = RelationGetIndexList(rel); + + foreach_oid(indexoid, indexoidlist) { - if (rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE) - { - PartitionDesc partdesc; + HeapTuple indexTuple; + Form_pg_index indexStruct; - partdesc = RelationGetPartitionDesc(rel, true); + indexTuple = SearchSysCache1(INDEXRELID, ObjectIdGetDatum(indexoid)); + if (!HeapTupleIsValid(indexTuple)) + elog(ERROR, "cache lookup failed for index %u", indexoid); + indexStruct = (Form_pg_index) GETSTRUCT(indexTuple); - if (partdesc->nparts > 0) - ereport(ERROR, - errcode(ERRCODE_INVALID_TABLE_DEFINITION), - errmsg("cannot remove constraint from only the partitioned table when partitions exist"), - errhint("Do not specify the ONLY keyword.")); - } - else if (rel->rd_rel->relhassubclass && - find_inheritance_children(RelationGetRelid(rel), NoLock) != NIL) + /* + * If the index is not a primary key or an index used as replica + * identity, skip the check. + */ + if (indexStruct->indisprimary || indexStruct->indisreplident) { - ereport(ERROR, - errcode(ERRCODE_INVALID_TABLE_DEFINITION), - errmsg("not-null constraint on column \"%s\" must be removed in child tables too", - colName), - errhint("Do not specify the ONLY keyword.")); + /* + * Loop over each attribute in the primary key or the index used + * as replica identity and see if it matches the to-be-altered + * attribute. + */ + for (int i = 0; i < indexStruct->indnkeyatts; i++) + { + if (indexStruct->indkey.values[i] == attnum) + { + if (indexStruct->indisprimary) + ereport(ERROR, + (errcode(ERRCODE_INVALID_TABLE_DEFINITION), + errmsg("column \"%s\" is in a primary key", + colName))); + else + ereport(ERROR, + (errcode(ERRCODE_INVALID_TABLE_DEFINITION), + errmsg("column \"%s\" is in index used as replica identity", + colName))); + } + } } + + ReleaseSysCache(indexTuple); } - /* - * If rel is partition, shouldn't drop NOT NULL if parent has the same. - */ + list_free(indexoidlist); + + /* If rel is partition, shouldn't drop NOT NULL if parent has the same */ if (rel->rd_rel->relispartition) { Oid parentId = get_partition_parent(RelationGetRelid(rel), false); @@ -7671,52 +7609,19 @@ ATExecDropNotNull(Relation rel, const char *colName, bool recurse, } /* - * Find the constraint that makes this column NOT NULL, and drop it if we - * see one. dropconstraint_internal() will do necessary consistency - * checking. If there isn't one, there are two possibilities: either the - * column is marked attnotnull because it's part of the primary key, and - * then we just throw an appropriate error; or it's a leftover marking - * that we can remove. However, before doing the latter, to avoid - * breaking consistency any further, prevent this if the column is part of - * the replica identity. + * Okay, actually perform the catalog change ... if needed */ - conTup = findNotNullConstraint(RelationGetRelid(rel), colName); - if (conTup == NULL) + if (attTup->attnotnull) { - Bitmapset *pkcols; - Bitmapset *ircols; - - /* - * If the column is in a primary key, throw a specific error message. - */ - pkcols = RelationGetIndexAttrBitmap(rel, INDEX_ATTR_BITMAP_PRIMARY_KEY); - if (bms_is_member(attnum - FirstLowInvalidHeapAttributeNumber, - pkcols)) - ereport(ERROR, - errcode(ERRCODE_INVALID_TABLE_DEFINITION), - errmsg("column \"%s\" is in a primary key", colName)); - - /* Also throw an error if the column is in the replica identity */ - ircols = RelationGetIndexAttrBitmap(rel, INDEX_ATTR_BITMAP_IDENTITY_KEY); - if (bms_is_member(attnum - FirstLowInvalidHeapAttributeNumber, ircols)) - ereport(ERROR, - errcode(ERRCODE_INVALID_TABLE_DEFINITION), - errmsg("column \"%s\" is in index used as replica identity", - get_attname(RelationGetRelid(rel), attnum, false))); - - /* Otherwise, just remove the attnotnull marking and do nothing else. */ attTup->attnotnull = false; + CatalogTupleUpdate(attr_rel, &tuple->t_self, tuple); - } - else - { - /* The normal case: we have a pg_constraint row, remove it */ - readyRels = NIL; - dropconstraint_internal(rel, conTup, DROP_RESTRICT, recurse, false, - false, &readyRels, lockmode); - heap_freetuple(conTup); + ObjectAddressSubSet(address, RelationRelationId, + RelationGetRelid(rel), attnum); } + else + address = InvalidObjectAddress; InvokeObjectPostAlterHook(RelationRelationId, RelationGetRelid(rel), attnum); @@ -7727,147 +7632,102 @@ ATExecDropNotNull(Relation rel, const char *colName, bool recurse, } /* - * Helper to set pg_attribute.attnotnull if it isn't set, and to tell phase 3 - * to verify it; recurses to apply the same to children. - * - * When called to alter an existing table, 'wqueue' must be given so that we can - * queue a check that existing tuples pass the constraint. When called from - * table creation, 'wqueue' should be passed as NULL. - * - * Returns true if the flag was set in any table, otherwise false. + * ALTER TABLE ALTER COLUMN SET NOT NULL */ -static bool -set_attnotnull(List **wqueue, Relation rel, AttrNumber attnum, bool recurse, - LOCKMODE lockmode) -{ - HeapTuple tuple; - Form_pg_attribute attForm; - bool retval = false; - /* Guard against stack overflow due to overly deep inheritance tree. */ - check_stack_depth(); +static void +ATPrepSetNotNull(List **wqueue, Relation rel, + AlterTableCmd *cmd, bool recurse, bool recursing, + LOCKMODE lockmode, AlterTableUtilityContext *context) +{ + /* + * If we're already recursing, there's nothing to do; the topmost + * invocation of ATSimpleRecursion already visited all children. + */ + if (recursing) + return; - tuple = SearchSysCacheCopyAttNum(RelationGetRelid(rel), attnum); - if (!HeapTupleIsValid(tuple)) - elog(ERROR, "cache lookup failed for attribute %d of relation %u", - attnum, RelationGetRelid(rel)); - attForm = (Form_pg_attribute) GETSTRUCT(tuple); - if (!attForm->attnotnull) + /* + * If the target column is already marked NOT NULL, we can skip recursing + * to children, because their columns should already be marked NOT NULL as + * well. But there's no point in checking here unless the relation has + * some children; else we can just wait till execution to check. (If it + * does have children, however, this can save taking per-child locks + * unnecessarily. This greatly improves concurrency in some parallel + * restore scenarios.) + * + * Unfortunately, we can only apply this optimization to partitioned + * tables, because traditional inheritance doesn't enforce that child + * columns be NOT NULL when their parent is. (That's a bug that should + * get fixed someday.) + */ + if (rel->rd_rel->relhassubclass && + rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE) { - Relation attr_rel; - - attr_rel = table_open(AttributeRelationId, RowExclusiveLock); - - attForm->attnotnull = true; - CatalogTupleUpdate(attr_rel, &tuple->t_self, tuple); - - table_close(attr_rel, RowExclusiveLock); + HeapTuple tuple; + bool attnotnull; - /* - * And set up for existing values to be checked, unless another - * constraint already proves this. - */ - if (wqueue && !NotNullImpliedByRelConstraints(rel, attForm)) - { - AlteredTableInfo *tab; + tuple = SearchSysCacheAttName(RelationGetRelid(rel), cmd->name); - tab = ATGetQueueEntry(wqueue, rel); - tab->verify_new_notnull = true; - } + /* Might as well throw the error now, if name is bad */ + if (!HeapTupleIsValid(tuple)) + ereport(ERROR, + (errcode(ERRCODE_UNDEFINED_COLUMN), + errmsg("column \"%s\" of relation \"%s\" does not exist", + cmd->name, RelationGetRelationName(rel)))); - retval = true; + attnotnull = ((Form_pg_attribute) GETSTRUCT(tuple))->attnotnull; + ReleaseSysCache(tuple); + if (attnotnull) + return; } - if (recurse) + /* + * If we have ALTER TABLE ONLY ... SET NOT NULL on a partitioned table, + * apply ALTER TABLE ... CHECK NOT NULL to every child. Otherwise, use + * normal recursion logic. + */ + if (rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE && + !recurse) { - List *children; - ListCell *lc; - - /* Make above update visible, for multiple inheritance cases */ - if (retval) - CommandCounterIncrement(); - - children = find_inheritance_children(RelationGetRelid(rel), lockmode); - foreach(lc, children) - { - Oid childrelid = lfirst_oid(lc); - Relation childrel; - AttrNumber childattno; - - /* find_inheritance_children already got lock */ - childrel = table_open(childrelid, NoLock); - CheckTableNotInUse(childrel, "ALTER TABLE"); + AlterTableCmd *newcmd = makeNode(AlterTableCmd); - childattno = get_attnum(RelationGetRelid(childrel), - get_attname(RelationGetRelid(rel), attnum, - false)); - retval |= set_attnotnull(wqueue, childrel, childattno, - recurse, lockmode); - table_close(childrel, NoLock); - } + newcmd->subtype = AT_CheckNotNull; + newcmd->name = pstrdup(cmd->name); + ATSimpleRecursion(wqueue, rel, newcmd, true, lockmode, context); } - - return retval; + else + ATSimpleRecursion(wqueue, rel, cmd, recurse, lockmode, context); } /* - * ALTER TABLE ALTER COLUMN SET NOT NULL - * - * Add a not-null constraint to a single table and its children. Returns - * the address of the constraint added to the parent relation, if one gets - * added, or InvalidObjectAddress otherwise. - * - * We must recurse to child tables during execution, rather than using - * ALTER TABLE's normal prep-time recursion. + * Return the address of the modified column. If the column was already NOT + * NULL, InvalidObjectAddress is returned. */ static ObjectAddress -ATExecSetNotNull(List **wqueue, Relation rel, char *conName, char *colName, - bool recurse, bool recursing, List **readyRels, - LOCKMODE lockmode) +ATExecSetNotNull(AlteredTableInfo *tab, Relation rel, + const char *colName, LOCKMODE lockmode) { HeapTuple tuple; - Relation constr_rel; - ScanKeyData skey; - SysScanDesc conscan; AttrNumber attnum; + Relation attr_rel; ObjectAddress address; - Constraint *constraint; - CookedConstraint *ccon; - List *cooked; - bool is_no_inherit = false; - List *ready = NIL; - - /* Guard against stack overflow due to overly deep inheritance tree. */ - check_stack_depth(); /* - * In cases of multiple inheritance, we might visit the same child more - * than once. In the topmost call, set up a list that we fill with all - * visited relations, to skip those. + * lookup the attribute */ - if (readyRels == NULL) - { - Assert(!recursing); - readyRels = &ready; - } - if (list_member_oid(*readyRels, RelationGetRelid(rel))) - return InvalidObjectAddress; - *readyRels = lappend_oid(*readyRels, RelationGetRelid(rel)); + attr_rel = table_open(AttributeRelationId, RowExclusiveLock); - /* At top level, permission check was done in ATPrepCmd, else do it */ - if (recursing) - { - ATSimplePermissions(AT_AddConstraint, rel, ATT_TABLE | ATT_FOREIGN_TABLE); - Assert(conName != NULL); - } + tuple = SearchSysCacheCopyAttName(RelationGetRelid(rel), colName); - attnum = get_attnum(RelationGetRelid(rel), colName); - if (attnum == InvalidAttrNumber) + if (!HeapTupleIsValid(tuple)) ereport(ERROR, (errcode(ERRCODE_UNDEFINED_COLUMN), errmsg("column \"%s\" of relation \"%s\" does not exist", colName, RelationGetRelationName(rel)))); + attnum = ((Form_pg_attribute) GETSTRUCT(tuple))->attnum; + /* Prevent them from altering a system attribute */ if (attnum <= 0) ereport(ERROR, @@ -7875,188 +7735,80 @@ ATExecSetNotNull(List **wqueue, Relation rel, char *conName, char *colName, errmsg("cannot alter system column \"%s\"", colName))); - /* See if there's already a constraint */ - constr_rel = table_open(ConstraintRelationId, RowExclusiveLock); - ScanKeyInit(&skey, - Anum_pg_constraint_conrelid, - BTEqualStrategyNumber, F_OIDEQ, - ObjectIdGetDatum(RelationGetRelid(rel))); - conscan = systable_beginscan(constr_rel, ConstraintRelidTypidNameIndexId, true, - NULL, 1, &skey); - - while (HeapTupleIsValid(tuple = systable_getnext(conscan))) + /* + * Okay, actually perform the catalog change ... if needed + */ + if (!((Form_pg_attribute) GETSTRUCT(tuple))->attnotnull) { - Form_pg_constraint conForm = (Form_pg_constraint) GETSTRUCT(tuple); - bool changed = false; - HeapTuple copytup; - - if (conForm->contype != CONSTRAINT_NOTNULL) - continue; + ((Form_pg_attribute) GETSTRUCT(tuple))->attnotnull = true; - if (extractNotNullColumn(tuple) != attnum) - continue; - - copytup = heap_copytuple(tuple); - conForm = (Form_pg_constraint) GETSTRUCT(copytup); - - /* - * Don't let a NO INHERIT constraint be changed into inherit. - */ - if (conForm->connoinherit && recurse) - ereport(ERROR, - errcode(ERRCODE_FEATURE_NOT_SUPPORTED), - errmsg("cannot change NO INHERIT status of NOT NULL constraint \"%s\" on relation \"%s\"", - NameStr(conForm->conname), - RelationGetRelationName(rel))); + CatalogTupleUpdate(attr_rel, &tuple->t_self, tuple); /* - * If we find an appropriate constraint, we're almost done, but just - * need to change some properties on it: if we're recursing, increment - * coninhcount; if not, set conislocal if not already set. + * Ordinarily phase 3 must ensure that no NULLs exist in columns that + * are set NOT NULL; however, if we can find a constraint which proves + * this then we can skip that. We needn't bother looking if we've + * already found that we must verify some other not-null constraint. */ - if (recursing) - { - conForm->coninhcount++; - changed = true; - } - else if (!conForm->conislocal) + if (!tab->verify_new_notnull && + !NotNullImpliedByRelConstraints(rel, (Form_pg_attribute) GETSTRUCT(tuple))) { - conForm->conislocal = true; - changed = true; - } - - if (changed) - { - CatalogTupleUpdate(constr_rel, ©tup->t_self, copytup); - ObjectAddressSet(address, ConstraintRelationId, conForm->oid); + /* Tell Phase 3 it needs to test the constraint */ + tab->verify_new_notnull = true; } - systable_endscan(conscan); - table_close(constr_rel, RowExclusiveLock); - - if (changed) - return address; - else - return InvalidObjectAddress; - } - - systable_endscan(conscan); - table_close(constr_rel, RowExclusiveLock); - - /* - * If we're asked not to recurse, and children exist, raise an error for - * partitioned tables. For inheritance, we act as if NO INHERIT had been - * specified. - */ - if (!recurse && - find_inheritance_children(RelationGetRelid(rel), - NoLock) != NIL) - { - if (rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE) - ereport(ERROR, - errcode(ERRCODE_INVALID_TABLE_DEFINITION), - errmsg("constraint must be added to child tables too"), - errhint("Do not specify the ONLY keyword.")); - else - is_no_inherit = true; + ObjectAddressSubSet(address, RelationRelationId, + RelationGetRelid(rel), attnum); } - - /* - * No constraint exists; we must add one. First determine a name to use, - * if we haven't already. - */ - if (!recursing) - { - Assert(conName == NULL); - conName = ChooseConstraintName(RelationGetRelationName(rel), - colName, "not_null", - RelationGetNamespace(rel), - NIL); - } - constraint = makeNode(Constraint); - constraint->contype = CONSTR_NOTNULL; - constraint->conname = conName; - constraint->deferrable = false; - constraint->initdeferred = false; - constraint->location = -1; - constraint->keys = list_make1(makeString(colName)); - constraint->is_no_inherit = is_no_inherit; - constraint->inhcount = recursing ? 1 : 0; - constraint->skip_validation = false; - constraint->initially_valid = true; - - /* and do it */ - cooked = AddRelationNewConstraints(rel, NIL, list_make1(constraint), - false, !recursing, false, NULL); - ccon = linitial(cooked); - ObjectAddressSet(address, ConstraintRelationId, ccon->conoid); + else + address = InvalidObjectAddress; InvokeObjectPostAlterHook(RelationRelationId, RelationGetRelid(rel), attnum); - /* - * Mark pg_attribute.attnotnull for the column. Tell that function not to - * recurse, because we're going to do it here. - */ - set_attnotnull(wqueue, rel, attnum, false, lockmode); - - /* - * Recurse to propagate the constraint to children that don't have one. - */ - if (recurse) - { - List *children; - ListCell *lc; - - children = find_inheritance_children(RelationGetRelid(rel), - lockmode); - - foreach(lc, children) - { - Relation childrel; - - childrel = table_open(lfirst_oid(lc), NoLock); - - ATExecSetNotNull(wqueue, childrel, - conName, colName, recurse, true, - readyRels, lockmode); - - table_close(childrel, NoLock); - } - } + table_close(attr_rel, RowExclusiveLock); return address; } /* - * ALTER TABLE ALTER COLUMN SET ATTNOTNULL + * ALTER TABLE ALTER COLUMN CHECK NOT NULL * - * This doesn't exist in the grammar; it's used when creating a - * primary key and the column is not already marked attnotnull. + * This doesn't exist in the grammar, but we generate AT_CheckNotNull + * commands against the partitions of a partitioned table if the user + * writes ALTER TABLE ONLY ... SET NOT NULL on the partitioned table, + * or tries to create a primary key on it (which internally creates + * AT_SetNotNull on the partitioned table). Such a command doesn't + * allow us to actually modify any partition, but we want to let it + * go through if the partitions are already properly marked. + * + * In future, this might need to adjust the child table's state, likely + * by incrementing an inheritance count for the attnotnull constraint. + * For now we need only check for the presence of the flag. */ -static ObjectAddress -ATExecSetAttNotNull(List **wqueue, Relation rel, - const char *colName, LOCKMODE lockmode) +static void +ATExecCheckNotNull(AlteredTableInfo *tab, Relation rel, + const char *colName, LOCKMODE lockmode) { - AttrNumber attnum; - ObjectAddress address = InvalidObjectAddress; + HeapTuple tuple; - attnum = get_attnum(RelationGetRelid(rel), colName); - if (attnum == InvalidAttrNumber) + tuple = SearchSysCacheAttName(RelationGetRelid(rel), colName); + + if (!HeapTupleIsValid(tuple)) ereport(ERROR, errcode(ERRCODE_UNDEFINED_COLUMN), errmsg("column \"%s\" of relation \"%s\" does not exist", colName, RelationGetRelationName(rel))); - /* - * Make the change, if necessary, and only if so report the column as - * changed - */ - if (set_attnotnull(wqueue, rel, attnum, false, lockmode)) - ObjectAddressSubSet(address, RelationRelationId, - RelationGetRelid(rel), attnum); + if (!((Form_pg_attribute) GETSTRUCT(tuple))->attnotnull) + ereport(ERROR, + (errcode(ERRCODE_INVALID_TABLE_DEFINITION), + errmsg("constraint must be added to child tables too"), + errdetail("Column \"%s\" of relation \"%s\" is not already NOT NULL.", + colName, RelationGetRelationName(rel)), + errhint("Do not specify the ONLY keyword."))); - return address; + ReleaseSysCache(tuple); } /* @@ -9363,85 +9115,6 @@ ATExecDropColumn(List **wqueue, Relation rel, const char *colName, } /* - * Prepare to add a primary key on an inheritance parent, by adding NOT NULL - * constraint on its children. - */ -static void -ATPrepAddPrimaryKey(List **wqueue, Relation rel, AlterTableCmd *cmd, - LOCKMODE lockmode, AlterTableUtilityContext *context) -{ - List *children; - List *newconstrs = NIL; - IndexStmt *indexstmt; - - /* No work if not creating a primary key */ - if (!IsA(cmd->def, IndexStmt)) - return; - indexstmt = castNode(IndexStmt, cmd->def); - if (!indexstmt->primary) - return; - - /* No work if no legacy inheritance children are present */ - if (rel->rd_rel->relkind != RELKIND_RELATION || - !rel->rd_rel->relhassubclass) - return; - - /* - * Acquire locks all the way down the hierarchy. The recursion to lower - * levels occurs at execution time as necessary, so we don't need to do it - * here, and we don't need the returned list either. - */ - (void) find_all_inheritors(RelationGetRelid(rel), lockmode, NULL); - - /* - * Construct the list of constraints that we need to add to each child - * relation. - */ - foreach_node(IndexElem, elem, indexstmt->indexParams) - { - Constraint *nnconstr; - - Assert(elem->expr == NULL); - - nnconstr = makeNode(Constraint); - nnconstr->contype = CONSTR_NOTNULL; - nnconstr->conname = NULL; /* XXX use PK name? */ - nnconstr->inhcount = 1; - nnconstr->deferrable = false; - nnconstr->initdeferred = false; - nnconstr->location = -1; - nnconstr->keys = list_make1(makeString(elem->name)); - nnconstr->skip_validation = false; - nnconstr->initially_valid = true; - - newconstrs = lappend(newconstrs, nnconstr); - } - - /* Finally, add AT subcommands to add each constraint to each child. */ - children = find_inheritance_children(RelationGetRelid(rel), NoLock); - foreach_oid(childrelid, children) - { - Relation childrel = table_open(childrelid, NoLock); - AlterTableCmd *newcmd = makeNode(AlterTableCmd); - ListCell *lc2; - - newcmd->subtype = AT_AddConstraint; - newcmd->recurse = true; - - foreach(lc2, newconstrs) - { - /* ATPrepCmd copies newcmd, so we can scribble on it here */ - newcmd->def = lfirst(lc2); - - ATPrepCmd(wqueue, childrel, newcmd, - true, false, lockmode, context); - } - - table_close(childrel, NoLock); - } -} - -/* * ALTER TABLE ADD INDEX * * There is no such command in the grammar, but parse_utilcmd.c converts @@ -9636,18 +9309,17 @@ ATExecAddConstraint(List **wqueue, AlteredTableInfo *tab, Relation rel, Assert(IsA(newConstraint, Constraint)); /* - * Currently, we only expect to see CONSTR_CHECK, CONSTR_NOTNULL and - * CONSTR_FOREIGN nodes arriving here (see the preprocessing done in - * parse_utilcmd.c). + * Currently, we only expect to see CONSTR_CHECK and CONSTR_FOREIGN nodes + * arriving here (see the preprocessing done in parse_utilcmd.c). Use a + * switch anyway to make it easier to add more code later. */ switch (newConstraint->contype) { case CONSTR_CHECK: - case CONSTR_NOTNULL: address = - ATAddCheckNNConstraint(wqueue, tab, rel, - newConstraint, recurse, false, is_readd, - lockmode); + ATAddCheckConstraint(wqueue, tab, rel, + newConstraint, recurse, false, is_readd, + lockmode); break; case CONSTR_FOREIGN: @@ -9728,9 +9400,9 @@ ChooseForeignKeyConstraintNameAddition(List *colnames) } /* - * Add a check or not-null constraint to a single table and its children. - * Returns the address of the constraint added to the parent relation, - * if one gets added, or InvalidObjectAddress otherwise. + * Add a check constraint to a single table and its children. Returns the + * address of the constraint added to the parent relation, if one gets added, + * or InvalidObjectAddress otherwise. * * Subroutine for ATExecAddConstraint. * @@ -9743,9 +9415,9 @@ ChooseForeignKeyConstraintNameAddition(List *colnames) * the parent table and pass that down. */ static ObjectAddress -ATAddCheckNNConstraint(List **wqueue, AlteredTableInfo *tab, Relation rel, - Constraint *constr, bool recurse, bool recursing, - bool is_readd, LOCKMODE lockmode) +ATAddCheckConstraint(List **wqueue, AlteredTableInfo *tab, Relation rel, + Constraint *constr, bool recurse, bool recursing, + bool is_readd, LOCKMODE lockmode) { List *newcons; ListCell *lcon; @@ -9753,9 +9425,6 @@ ATAddCheckNNConstraint(List **wqueue, AlteredTableInfo *tab, Relation rel, ListCell *child; ObjectAddress address = InvalidObjectAddress; - /* Guard against stack overflow due to overly deep inheritance tree. */ - check_stack_depth(); - /* At top level, permission check was done in ATPrepCmd, else do it */ if (recursing) ATSimplePermissions(AT_AddConstraint, rel, ATT_TABLE | ATT_FOREIGN_TABLE); @@ -9786,7 +9455,7 @@ ATAddCheckNNConstraint(List **wqueue, AlteredTableInfo *tab, Relation rel, { CookedConstraint *ccon = (CookedConstraint *) lfirst(lcon); - if (!ccon->skip_validation && ccon->contype != CONSTR_NOTNULL) + if (!ccon->skip_validation) { NewConstraint *newcon; @@ -9802,19 +9471,11 @@ ATAddCheckNNConstraint(List **wqueue, AlteredTableInfo *tab, Relation rel, if (constr->conname == NULL) constr->conname = ccon->name; - /* - * If adding a not-null constraint, set the pg_attribute flag and tell - * phase 3 to verify existing rows, if needed. - */ - if (constr->contype == CONSTR_NOTNULL) - set_attnotnull(wqueue, rel, ccon->attnum, - !ccon->is_no_inherit, lockmode); - ObjectAddressSet(address, ConstraintRelationId, ccon->conoid); } /* At this point we must have a locked-down name to use */ - Assert(newcons == NIL || constr->conname != NULL); + Assert(constr->conname != NULL); /* Advance command counter in case same table is visited multiple times */ CommandCounterIncrement(); @@ -9852,12 +9513,6 @@ ATAddCheckNNConstraint(List **wqueue, AlteredTableInfo *tab, Relation rel, (errcode(ERRCODE_INVALID_TABLE_DEFINITION), errmsg("constraint must be added to child tables too"))); - /* - * The constraint must appear as inherited in children, so create a - * modified constraint object to use. - */ - constr = copyObject(constr); - constr->inhcount = 1; foreach(child, children) { Oid childrelid = lfirst_oid(child); @@ -9871,13 +9526,9 @@ ATAddCheckNNConstraint(List **wqueue, AlteredTableInfo *tab, Relation rel, /* Find or create work queue entry for this table */ childtab = ATGetQueueEntry(wqueue, childrel); - /* - * Recurse to child. XXX if we didn't create a constraint on the - * parent because it already existed, and we do create one on a child, - * should we return that child's constraint ObjectAddress here? - */ - ATAddCheckNNConstraint(wqueue, childtab, childrel, - constr, recurse, true, is_readd, lockmode); + /* Recurse to child */ + ATAddCheckConstraint(wqueue, childtab, childrel, + constr, recurse, true, is_readd, lockmode); table_close(childrel, NoLock); } @@ -12889,14 +12540,23 @@ createForeignKeyCheckTriggers(Oid myRelOid, Oid refRelOid, */ static void ATExecDropConstraint(Relation rel, const char *constrName, - DropBehavior behavior, bool recurse, + DropBehavior behavior, + bool recurse, bool recursing, bool missing_ok, LOCKMODE lockmode) { + List *children; Relation conrel; + Form_pg_constraint con; SysScanDesc scan; ScanKeyData skey[3]; HeapTuple tuple; bool found = false; + bool is_no_inherit_constraint = false; + char contype; + + /* At top level, permission check was done in ATPrepCmd, else do it */ + if (recursing) + ATSimplePermissions(AT_DropConstraint, rel, ATT_TABLE | ATT_FOREIGN_TABLE); conrel = table_open(ConstraintRelationId, RowExclusiveLock); @@ -12921,260 +12581,80 @@ ATExecDropConstraint(Relation rel, const char *constrName, /* There can be at most one matching row */ if (HeapTupleIsValid(tuple = systable_getnext(scan))) { - List *readyRels = NIL; + ObjectAddress conobj; - dropconstraint_internal(rel, tuple, behavior, recurse, false, - missing_ok, &readyRels, lockmode); - found = true; - } + con = (Form_pg_constraint) GETSTRUCT(tuple); - systable_endscan(scan); - - if (!found) - { - if (!missing_ok) + /* Don't drop inherited constraints */ + if (con->coninhcount > 0 && !recursing) ereport(ERROR, - errcode(ERRCODE_UNDEFINED_OBJECT), - errmsg("constraint \"%s\" of relation \"%s\" does not exist", - constrName, RelationGetRelationName(rel))); - else - ereport(NOTICE, - errmsg("constraint \"%s\" of relation \"%s\" does not exist, skipping", - constrName, RelationGetRelationName(rel))); - } - - table_close(conrel, RowExclusiveLock); -} - -/* - * Remove a constraint, using its pg_constraint tuple - * - * Implementation for ALTER TABLE DROP CONSTRAINT and ALTER TABLE ALTER COLUMN - * DROP NOT NULL. - * - * Returns the address of the constraint being removed. - */ -static ObjectAddress -dropconstraint_internal(Relation rel, HeapTuple constraintTup, DropBehavior behavior, - bool recurse, bool recursing, bool missing_ok, List **readyRels, - LOCKMODE lockmode) -{ - Relation conrel; - Form_pg_constraint con; - ObjectAddress conobj; - List *children; - bool is_no_inherit_constraint = false; - char *constrName; - List *unconstrained_cols = NIL; - char *colname = NULL; - bool dropping_pk = false; - - if (list_member_oid(*readyRels, RelationGetRelid(rel))) - return InvalidObjectAddress; - *readyRels = lappend_oid(*readyRels, RelationGetRelid(rel)); - - /* Guard against stack overflow due to overly deep inheritance tree. */ - check_stack_depth(); - - /* At top level, permission check was done in ATPrepCmd, else do it */ - if (recursing) - ATSimplePermissions(AT_DropConstraint, rel, ATT_TABLE | ATT_FOREIGN_TABLE); - - conrel = table_open(ConstraintRelationId, RowExclusiveLock); - - con = (Form_pg_constraint) GETSTRUCT(constraintTup); - constrName = NameStr(con->conname); - - /* - * If we're asked to drop a constraint which is both defined locally and - * inherited, we can simply mark it as no longer having a local - * definition, and no further changes are required. - * - * XXX We do this for not-null constraints only, not CHECK, because the - * latter have historically not behaved this way and it might be confusing - * to change the behavior now. - */ - if (con->contype == CONSTRAINT_NOTNULL && - con->conislocal && con->coninhcount > 0) - { - HeapTuple copytup; - - copytup = heap_copytuple(constraintTup); - con = (Form_pg_constraint) GETSTRUCT(copytup); - con->conislocal = false; - CatalogTupleUpdate(conrel, ©tup->t_self, copytup); - ObjectAddressSet(conobj, ConstraintRelationId, con->oid); - - CommandCounterIncrement(); - table_close(conrel, RowExclusiveLock); - return conobj; - } - - /* Don't allow drop of 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)))); - - /* - * See if we have a not-null constraint or a PRIMARY KEY. If so, we have - * more checks and actions below, so obtain the list of columns that are - * constrained by the constraint being dropped. - */ - if (con->contype == CONSTRAINT_NOTNULL) - { - AttrNumber colnum; - - colnum = extractNotNullColumn(constraintTup); - unconstrained_cols = list_make1_int(colnum); - colname = NameStr(TupleDescAttr(RelationGetDescr(rel), - colnum - 1)->attname); - } - else if (con->contype == CONSTRAINT_PRIMARY) - { - Datum adatum; - ArrayType *arr; - int numkeys; - bool isNull; - int16 *attnums; + (errcode(ERRCODE_INVALID_TABLE_DEFINITION), + errmsg("cannot drop inherited constraint \"%s\" of relation \"%s\"", + constrName, RelationGetRelationName(rel)))); - dropping_pk = true; + is_no_inherit_constraint = con->connoinherit; + contype = con->contype; - adatum = heap_getattr(constraintTup, Anum_pg_constraint_conkey, - RelationGetDescr(conrel), &isNull); - if (isNull) - elog(ERROR, "null conkey for constraint %u", con->oid); - arr = DatumGetArrayTypeP(adatum); /* ensure not toasted */ - numkeys = ARR_DIMS(arr)[0]; - if (ARR_NDIM(arr) != 1 || - numkeys < 0 || - ARR_HASNULL(arr) || - ARR_ELEMTYPE(arr) != INT2OID) - elog(ERROR, "conkey is not a 1-D smallint array"); - attnums = (int16 *) ARR_DATA_PTR(arr); + /* + * If it's a foreign-key constraint, we'd better lock the referenced + * table and check that that's not in use, just as we've already done + * for the constrained table (else we might, eg, be dropping a trigger + * that has unfired events). But we can/must skip that in the + * self-referential case. + */ + if (contype == CONSTRAINT_FOREIGN && + con->confrelid != RelationGetRelid(rel)) + { + Relation frel; - for (int i = 0; i < numkeys; i++) - unconstrained_cols = lappend_int(unconstrained_cols, attnums[i]); - } + /* Must match lock taken by RemoveTriggerById: */ + frel = table_open(con->confrelid, AccessExclusiveLock); + CheckTableNotInUse(frel, "ALTER TABLE"); + table_close(frel, NoLock); + } - is_no_inherit_constraint = con->connoinherit; + /* + * Perform the actual constraint deletion + */ + conobj.classId = ConstraintRelationId; + conobj.objectId = con->oid; + conobj.objectSubId = 0; - /* - * If it's a foreign-key constraint, we'd better lock the referenced table - * and check that that's not in use, just as we've already done for the - * constrained table (else we might, eg, be dropping a trigger that has - * unfired events). But we can/must skip that in the self-referential - * case. - */ - if (con->contype == CONSTRAINT_FOREIGN && - con->confrelid != RelationGetRelid(rel)) - { - Relation frel; + performDeletion(&conobj, behavior, 0); - /* Must match lock taken by RemoveTriggerById: */ - frel = table_open(con->confrelid, AccessExclusiveLock); - CheckTableNotInUse(frel, "ALTER TABLE"); - table_close(frel, NoLock); + found = true; } - /* - * Perform the actual constraint deletion - */ - ObjectAddressSet(conobj, ConstraintRelationId, con->oid); - performDeletion(&conobj, behavior, 0); + systable_endscan(scan); - /* - * If this was a NOT NULL or the primary key, verify that we still have - * constraints to support GENERATED AS IDENTITY or the replica identity. - */ - if (unconstrained_cols != NIL) + if (!found) { - Relation attrel; - Bitmapset *pkcols; - Bitmapset *ircols; - - /* Make implicit attnotnull changes visible */ - CommandCounterIncrement(); - - attrel = table_open(AttributeRelationId, RowExclusiveLock); - - /* - * We want to test columns for their presence in the primary key, but - * only if we're not dropping it. - */ - pkcols = dropping_pk ? NULL : - RelationGetIndexAttrBitmap(rel, - INDEX_ATTR_BITMAP_PRIMARY_KEY); - ircols = RelationGetIndexAttrBitmap(rel, INDEX_ATTR_BITMAP_IDENTITY_KEY); - - foreach_int(attnum, unconstrained_cols) + if (!missing_ok) { - HeapTuple atttup; - HeapTuple contup; - Form_pg_attribute attForm; - char attidentity; - - /* - * Obtain pg_attribute tuple and verify conditions on it. - */ - atttup = SearchSysCacheAttNum(RelationGetRelid(rel), attnum); - if (!HeapTupleIsValid(atttup)) - elog(ERROR, "cache lookup failed for attribute %d of relation %u", - attnum, RelationGetRelid(rel)); - attForm = (Form_pg_attribute) GETSTRUCT(atttup); - attidentity = attForm->attidentity; - ReleaseSysCache(atttup); - - /* - * Since the above deletion has been made visible, we can now - * search for any remaining constraints on this column (or these - * columns, in the case we're dropping a multicol primary key.) - * Then, verify whether any further NOT NULL or primary key - * exists: if none and this is a generated identity column or the - * replica identity, abort the whole thing. - */ - contup = findNotNullConstraintAttnum(RelationGetRelid(rel), attnum); - if (contup || - bms_is_member(attnum - FirstLowInvalidHeapAttributeNumber, - pkcols)) - continue; - - /* - * It's not valid to drop the not-null constraint for a GENERATED - * AS IDENTITY column. - */ - if (attidentity != '\0') - ereport(ERROR, - errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), - errmsg("column \"%s\" of relation \"%s\" is an identity column", - get_attname(RelationGetRelid(rel), attnum, - false), - RelationGetRelationName(rel))); - - /* - * It's not valid to drop the not-null constraint for a column in - * the replica identity index, either. (FULL is not affected.) - */ - if (bms_is_member(attnum - FirstLowInvalidHeapAttributeNumber, ircols)) - ereport(ERROR, - errcode(ERRCODE_INVALID_TABLE_DEFINITION), - errmsg("column \"%s\" is in index used as replica identity", - get_attname(RelationGetRelid(rel), attnum, false))); + ereport(ERROR, + (errcode(ERRCODE_UNDEFINED_OBJECT), + errmsg("constraint \"%s\" of relation \"%s\" does not exist", + constrName, RelationGetRelationName(rel)))); + } + else + { + ereport(NOTICE, + (errmsg("constraint \"%s\" of relation \"%s\" does not exist, skipping", + constrName, RelationGetRelationName(rel)))); + table_close(conrel, RowExclusiveLock); + return; } - table_close(attrel, RowExclusiveLock); } /* - * For partitioned tables, non-CHECK, non-NOT-NULL inherited constraints - * are dropped via the dependency mechanism, so we're done here. + * For partitioned tables, non-CHECK inherited constraints are dropped via + * the dependency mechanism, so we're done here. */ - if (con->contype != CONSTRAINT_CHECK && - con->contype != CONSTRAINT_NOTNULL && + if (contype != CONSTRAINT_CHECK && rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE) { table_close(conrel, RowExclusiveLock); - return conobj; + return; } /* @@ -13202,68 +12682,48 @@ dropconstraint_internal(Relation rel, HeapTuple constraintTup, DropBehavior beha foreach_oid(childrelid, children) { Relation childrel; - HeapTuple tuple; - Form_pg_constraint childcon; - - if (list_member_oid(*readyRels, childrelid)) - continue; /* child already processed */ + HeapTuple copy_tuple; /* find_inheritance_children already got lock */ childrel = table_open(childrelid, NoLock); CheckTableNotInUse(childrel, "ALTER TABLE"); - /* - * We search for not-null constraints by column name, and others by - * constraint name. - */ - if (con->contype == CONSTRAINT_NOTNULL) - { - tuple = findNotNullConstraint(childrelid, colname); - if (!HeapTupleIsValid(tuple)) - elog(ERROR, "cache lookup failed for not-null constraint on column \"%s\" of relation %u", - colname, RelationGetRelid(childrel)); - } - else - { - SysScanDesc scan; - ScanKeyData skey[3]; - - ScanKeyInit(&skey[0], - Anum_pg_constraint_conrelid, - BTEqualStrategyNumber, F_OIDEQ, - ObjectIdGetDatum(childrelid)); - ScanKeyInit(&skey[1], - Anum_pg_constraint_contypid, - BTEqualStrategyNumber, F_OIDEQ, - ObjectIdGetDatum(InvalidOid)); - ScanKeyInit(&skey[2], - Anum_pg_constraint_conname, - BTEqualStrategyNumber, F_NAMEEQ, - CStringGetDatum(constrName)); - scan = systable_beginscan(conrel, ConstraintRelidTypidNameIndexId, - true, NULL, 3, skey); - /* There can only be one, so no need to loop */ - tuple = systable_getnext(scan); - if (!HeapTupleIsValid(tuple)) - ereport(ERROR, - (errcode(ERRCODE_UNDEFINED_OBJECT), - errmsg("constraint \"%s\" of relation \"%s\" does not exist", - constrName, - RelationGetRelationName(childrel)))); - tuple = heap_copytuple(tuple); - systable_endscan(scan); - } + ScanKeyInit(&skey[0], + Anum_pg_constraint_conrelid, + BTEqualStrategyNumber, F_OIDEQ, + ObjectIdGetDatum(childrelid)); + ScanKeyInit(&skey[1], + Anum_pg_constraint_contypid, + BTEqualStrategyNumber, F_OIDEQ, + ObjectIdGetDatum(InvalidOid)); + ScanKeyInit(&skey[2], + Anum_pg_constraint_conname, + BTEqualStrategyNumber, F_NAMEEQ, + CStringGetDatum(constrName)); + scan = systable_beginscan(conrel, ConstraintRelidTypidNameIndexId, + true, NULL, 3, skey); + + /* There can be at most one matching row */ + if (!HeapTupleIsValid(tuple = systable_getnext(scan))) + ereport(ERROR, + (errcode(ERRCODE_UNDEFINED_OBJECT), + errmsg("constraint \"%s\" of relation \"%s\" does not exist", + constrName, + RelationGetRelationName(childrel)))); + + copy_tuple = heap_copytuple(tuple); + + systable_endscan(scan); - childcon = (Form_pg_constraint) GETSTRUCT(tuple); + con = (Form_pg_constraint) GETSTRUCT(copy_tuple); - /* Right now only CHECK and not-null constraints can be inherited */ - if (childcon->contype != CONSTRAINT_CHECK && - childcon->contype != CONSTRAINT_NOTNULL) - elog(ERROR, "inherited constraint is not a CHECK or not-null constraint"); + /* Right now only CHECK constraints can be inherited */ + if (con->contype != CONSTRAINT_CHECK) + elog(ERROR, "inherited constraint is not a CHECK constraint"); - if (childcon->coninhcount <= 0) /* shouldn't happen */ + if (con->coninhcount <= 0) /* shouldn't happen */ elog(ERROR, "relation %u has non-inherited constraint \"%s\"", - childrelid, NameStr(childcon->conname)); + childrelid, constrName); if (recurse) { @@ -13271,18 +12731,18 @@ dropconstraint_internal(Relation rel, HeapTuple constraintTup, DropBehavior beha * If the child constraint has other definition sources, just * decrement its inheritance count; if not, recurse to delete it. */ - if (childcon->coninhcount == 1 && !childcon->conislocal) + if (con->coninhcount == 1 && !con->conislocal) { /* Time to delete this child constraint, too */ - dropconstraint_internal(childrel, tuple, behavior, - recurse, true, missing_ok, readyRels, - lockmode); + ATExecDropConstraint(childrel, constrName, behavior, + true, true, + false, lockmode); } else { /* Child constraint must survive my deletion */ - childcon->coninhcount--; - CatalogTupleUpdate(conrel, &tuple->t_self, tuple); + con->coninhcount--; + CatalogTupleUpdate(conrel, ©_tuple->t_self, copy_tuple); /* Make update visible */ CommandCounterIncrement(); @@ -13291,91 +12751,25 @@ dropconstraint_internal(Relation rel, HeapTuple constraintTup, DropBehavior beha else { /* - * If we were told to drop ONLY in this table (no recursion) and - * there are no further parents for this constraint, we need to - * mark the inheritors' constraints as locally defined rather than - * inherited. + * 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. */ - childcon->coninhcount--; - if (childcon->coninhcount == 0) - childcon->conislocal = true; + con->coninhcount--; + con->conislocal = true; - CatalogTupleUpdate(conrel, &tuple->t_self, tuple); + CatalogTupleUpdate(conrel, ©_tuple->t_self, copy_tuple); /* Make update visible */ CommandCounterIncrement(); } - heap_freetuple(tuple); + heap_freetuple(copy_tuple); table_close(childrel, NoLock); } - /* - * In addition, when dropping a primary key from a legacy-inheritance - * parent table, we must recurse to children to mark the corresponding NOT - * NULL constraint as no longer inherited, or drop it if this its last - * reference. - */ - if (con->contype == CONSTRAINT_PRIMARY && - rel->rd_rel->relkind == RELKIND_RELATION && - rel->rd_rel->relhassubclass) - { - List *colnames = NIL; - List *pkready = NIL; - - /* - * Because primary keys are always marked as NO INHERIT, we don't have - * a list of children yet, so obtain one now. - */ - children = find_inheritance_children(RelationGetRelid(rel), lockmode); - - /* - * Find out the list of column names to process. Fortunately, we - * already have the list of column numbers. - */ - foreach_int(attnum, unconstrained_cols) - { - colnames = lappend(colnames, get_attname(RelationGetRelid(rel), - attnum, false)); - } - - foreach_oid(childrelid, children) - { - Relation childrel; - - if (list_member_oid(pkready, childrelid)) - continue; /* child already processed */ - - /* find_inheritance_children already got lock */ - childrel = table_open(childrelid, NoLock); - CheckTableNotInUse(childrel, "ALTER TABLE"); - - foreach_ptr(char, colName, colnames) - { - HeapTuple contup; - - contup = findNotNullConstraint(childrelid, colName); - if (contup == NULL) - elog(ERROR, "cache lookup failed for not-null constraint on column \"%s\", relation \"%s\"", - colName, RelationGetRelationName(childrel)); - - dropconstraint_internal(childrel, contup, - DROP_RESTRICT, true, true, - false, &pkready, - lockmode); - pkready = NIL; - } - - table_close(childrel, NoLock); - - pkready = lappend_oid(pkready, childrelid); - } - } - table_close(conrel, RowExclusiveLock); - - return conobj; } /* @@ -14479,10 +13873,9 @@ ATPostAlterTypeCleanup(List **wqueue, AlteredTableInfo *tab, LOCKMODE lockmode) /* * If the constraint is inherited (only), we don't want to inject a - * new definition here; it'll get recreated when - * ATAddCheckNNConstraint recurses from adding the parent table's - * constraint. But we had to carry the info this far so that we can - * drop the constraint below. + * new definition here; it'll get recreated when ATAddCheckConstraint + * recurses from adding the parent table's constraint. But we had to + * carry the info this far so that we can drop the constraint below. */ if (!conislocal) continue; @@ -14729,19 +14122,15 @@ ATPostAlterTypeParse(Oid oldId, Oid oldRelId, Oid refRelId, char *cmd, NIL, con->conname); } - else if (cmd->subtype == AT_SetAttNotNull) + else if (cmd->subtype == AT_SetNotNull) { /* - * We see this subtype when a primary key is created - * internally, for example when it is replaced with a new - * constraint (say because one of the columns changes - * type); in this case we need to reinstate attnotnull, - * because it was removed because of the drop of the old - * PK. Schedule this subcommand to an upcoming AT pass. + * The parser will create AT_SetNotNull subcommands for + * columns of PRIMARY KEY indexes/constraints, but we need + * not do anything with them here, because the columns' + * NOT NULL marks will already have been propagated into + * the new table definition. */ - cmd->subtype = AT_SetAttNotNull; - tab->subcmds[AT_PASS_OLD_COL_ATTRS] = - lappend(tab->subcmds[AT_PASS_OLD_COL_ATTRS], cmd); } else elog(ERROR, "unexpected statement subtype: %d", @@ -16316,13 +15705,6 @@ ATExecAddInherit(Relation child_rel, RangeVar *parent, LOCKMODE lockmode) /* OK to create inheritance */ CreateInheritance(child_rel, parent_rel, false); - /* - * If parent_rel has a primary key, then child_rel has not-null - * constraints that make these columns as non nullable. Make those - * constraints as inherited. - */ - ATInheritAdjustNotNulls(parent_rel, child_rel, 1); - ObjectAddressSet(address, RelationRelationId, RelationGetRelid(parent_rel)); @@ -16501,24 +15883,14 @@ MergeAttributesIntoExisting(Relation child_rel, Relation parent_rel, bool ispart RelationGetRelationName(child_rel), parent_attname))); /* - * If the parent has a not-null constraint that's not NO INHERIT, - * make sure the child has one too. - * - * Other constraints are checked elsewhere. + * Check child doesn't discard NOT NULL property. (Other + * constraints are checked elsewhere.) */ if (parent_att->attnotnull && !child_att->attnotnull) - { - HeapTuple contup; - - contup = findNotNullConstraintAttnum(RelationGetRelid(parent_rel), - parent_att->attnum); - if (HeapTupleIsValid(contup) && - !((Form_pg_constraint) GETSTRUCT(contup))->connoinherit) - ereport(ERROR, - errcode(ERRCODE_DATATYPE_MISMATCH), - errmsg("column \"%s\" in child table must be marked NOT NULL", - parent_attname)); - } + ereport(ERROR, + (errcode(ERRCODE_DATATYPE_MISMATCH), + errmsg("column \"%s\" in child table must be marked NOT NULL", + parent_attname))); /* * Child column must be generated if and only if parent column is. @@ -16619,8 +15991,7 @@ MergeConstraintsIntoExisting(Relation child_rel, Relation parent_rel) HeapTuple child_tuple; bool found = false; - if (parent_con->contype != CONSTRAINT_CHECK && - parent_con->contype != CONSTRAINT_NOTNULL) + if (parent_con->contype != CONSTRAINT_CHECK) continue; /* if the parent's constraint is marked NO INHERIT, it's not inherited */ @@ -16640,50 +16011,21 @@ MergeConstraintsIntoExisting(Relation child_rel, Relation parent_rel) Form_pg_constraint child_con = (Form_pg_constraint) GETSTRUCT(child_tuple); HeapTuple child_copy; - if (child_con->contype != parent_con->contype) + if (child_con->contype != CONSTRAINT_CHECK) continue; - /* - * CHECK constraint are matched by name, NOT NULL ones by - * attribute number - */ - if (child_con->contype == CONSTRAINT_CHECK) - { - if (strcmp(NameStr(parent_con->conname), - NameStr(child_con->conname)) != 0) - continue; - } - else if (child_con->contype == CONSTRAINT_NOTNULL) - { - AttrNumber parent_attno = extractNotNullColumn(parent_tuple); - AttrNumber child_attno = extractNotNullColumn(child_tuple); - - if (strcmp(get_attname(parent_relid, parent_attno, false), - get_attname(RelationGetRelid(child_rel), child_attno, - false)) != 0) - continue; - } + if (strcmp(NameStr(parent_con->conname), + NameStr(child_con->conname)) != 0) + continue; - if (child_con->contype == CONSTRAINT_CHECK && - !constraints_equivalent(parent_tuple, child_tuple, RelationGetDescr(constraintrel))) + if (!constraints_equivalent(parent_tuple, child_tuple, RelationGetDescr(constraintrel))) ereport(ERROR, (errcode(ERRCODE_DATATYPE_MISMATCH), errmsg("child table \"%s\" has different definition for check constraint \"%s\"", RelationGetRelationName(child_rel), NameStr(parent_con->conname)))); - /* - * If the CHECK child constraint is "no inherit" then cannot - * merge. - * - * This is not desirable for not-null constraints, mostly because - * it breaks our pg_upgrade strategy, but it also makes sense on - * its own: if a child has its own not-null constraint and then - * acquires a parent with the same constraint, then we start to - * enforce that constraint for all the descendants of that child - * too, if any. - */ - if (child_con->contype == CONSTRAINT_CHECK && - child_con->connoinherit) + /* If the child constraint is "no inherit" then cannot merge */ + if (child_con->connoinherit) ereport(ERROR, (errcode(ERRCODE_INVALID_OBJECT_DEFINITION), errmsg("constraint \"%s\" conflicts with non-inherited constraint on child table \"%s\"", @@ -16710,27 +16052,6 @@ MergeConstraintsIntoExisting(Relation child_rel, Relation parent_rel) ereport(ERROR, errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED), errmsg("too many inheritance parents")); - if (child_con->contype == CONSTRAINT_NOTNULL && - child_con->connoinherit) - { - /* - * If the child has children, it's not possible to turn a NO - * INHERIT constraint into an inheritable one: we would need - * to recurse to create constraints in those children, but - * this is not a good place to do that. - */ - if (child_rel->rd_rel->relhassubclass) - ereport(ERROR, - errmsg("cannot add NOT NULL constraint to column \"%s\" of relation \"%s\" with inheritance children", - get_attname(RelationGetRelid(child_rel), - extractNotNullColumn(child_tuple), - false), - RelationGetRelationName(child_rel)), - errdetail("Existing constraint \"%s\" is marked NO INHERIT.", - NameStr(child_con->conname))); - - child_con->connoinherit = false; - } /* * In case of partitions, an inherited constraint must be @@ -16753,20 +16074,10 @@ MergeConstraintsIntoExisting(Relation child_rel, Relation parent_rel) systable_endscan(child_scan); if (!found) - { - if (parent_con->contype == CONSTRAINT_NOTNULL) - ereport(ERROR, - errcode(ERRCODE_DATATYPE_MISMATCH), - errmsg("column \"%s\" in child table must be marked NOT NULL", - get_attname(parent_relid, - extractNotNullColumn(parent_tuple), - false))); - ereport(ERROR, (errcode(ERRCODE_DATATYPE_MISMATCH), errmsg("child table is missing constraint \"%s\"", NameStr(parent_con->conname)))); - } } systable_endscan(parent_scan); @@ -16804,18 +16115,6 @@ ATExecDropInherit(Relation rel, RangeVar *parent, LOCKMODE lockmode) /* Off to RemoveInheritance() where most of the work happens */ RemoveInheritance(rel, parent_rel, false); - /* - * If parent_rel has a primary key, then child_rel has not-null - * constraints that make these columns as non nullable. Mark those - * constraints as no longer inherited by this parent. - */ - ATInheritAdjustNotNulls(parent_rel, rel, -1); - - /* - * If the parent has a primary key, then we decrement counts for all NOT - * NULL constraints - */ - ObjectAddressSet(address, RelationRelationId, RelationGetRelid(parent_rel)); @@ -16924,7 +16223,6 @@ RemoveInheritance(Relation child_rel, Relation parent_rel, bool expect_detached) HeapTuple attributeTuple, constraintTuple; List *connames; - List *nncolumns; bool found; bool is_partitioning; @@ -16993,8 +16291,6 @@ RemoveInheritance(Relation child_rel, Relation parent_rel, bool expect_detached) * 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.) - * - * For NOT NULL columns, we store column numbers to match. */ catalogRelation = table_open(ConstraintRelationId, RowExclusiveLock); ScanKeyInit(&key[0], @@ -17005,7 +16301,6 @@ RemoveInheritance(Relation child_rel, Relation parent_rel, bool expect_detached) true, NULL, 1, key); connames = NIL; - nncolumns = NIL; while (HeapTupleIsValid(constraintTuple = systable_getnext(scan))) { @@ -17013,8 +16308,6 @@ RemoveInheritance(Relation child_rel, Relation parent_rel, bool expect_detached) if (con->contype == CONSTRAINT_CHECK) connames = lappend(connames, pstrdup(NameStr(con->conname))); - if (con->contype == CONSTRAINT_NOTNULL) - nncolumns = lappend_int(nncolumns, extractNotNullColumn(constraintTuple)); } systable_endscan(scan); @@ -17030,40 +16323,20 @@ RemoveInheritance(Relation child_rel, Relation parent_rel, bool expect_detached) while (HeapTupleIsValid(constraintTuple = systable_getnext(scan))) { Form_pg_constraint con = (Form_pg_constraint) GETSTRUCT(constraintTuple); - bool match = false; - ListCell *lc; + bool match; - /* - * Match CHECK constraints by name, not-null constraints by column - * number, and ignore all others. - */ - if (con->contype == CONSTRAINT_CHECK) - { - foreach(lc, connames) - { - if (con->contype == CONSTRAINT_CHECK && - strcmp(NameStr(con->conname), (char *) lfirst(lc)) == 0) - { - match = true; - break; - } - } - } - else if (con->contype == CONSTRAINT_NOTNULL) - { - AttrNumber child_attno = extractNotNullColumn(constraintTuple); + if (con->contype != CONSTRAINT_CHECK) + continue; - foreach(lc, nncolumns) + match = false; + foreach_ptr(char, chkname, connames) + { + if (strcmp(NameStr(con->conname), chkname) == 0) { - if (lfirst_int(lc) == child_attno) - { - match = true; - break; - } + match = true; + break; } } - else - continue; if (match) { @@ -17103,54 +16376,6 @@ RemoveInheritance(Relation child_rel, Relation parent_rel, bool expect_detached) } /* - * Adjust coninhcount of not-null constraints upwards or downwards when a - * table is marked as inheriting or no longer doing so a table with a primary - * key. - * - * Note: these constraints are not dropped, even if their inhcount goes to zero - * and conislocal is false. Instead we mark the constraints as locally defined. - * This is seen as more useful behavior, with no downsides. The user can always - * drop them afterwards. - */ -static void -ATInheritAdjustNotNulls(Relation parent_rel, Relation child_rel, int inhcount) -{ - Bitmapset *pkattnos; - - /* Quick exit when parent has no PK */ - if (!parent_rel->rd_rel->relhasindex) - return; - - pkattnos = RelationGetIndexAttrBitmap(parent_rel, - INDEX_ATTR_BITMAP_PRIMARY_KEY); - if (pkattnos != NULL) - { - Bitmapset *childattnums = NULL; - AttrMap *attmap; - int i; - - attmap = build_attrmap_by_name(RelationGetDescr(parent_rel), - RelationGetDescr(child_rel), true); - - i = -1; - while ((i = bms_next_member(pkattnos, i)) >= 0) - { - childattnums = bms_add_member(childattnums, - attmap->attnums[i + FirstLowInvalidHeapAttributeNumber - 1]); - } - - /* - * CCI is needed in case there's a NOT NULL PRIMARY KEY column in the - * parent: the relevant not-null constraint in the child already had - * its inhcount modified earlier. - */ - CommandCounterIncrement(); - AdjustNotNullInheritance(RelationGetRelid(child_rel), childattnums, - inhcount); - } -} - -/* * Drop the dependency created by StoreCatalogInheritance1 (CREATE TABLE * INHERITS/ALTER TABLE INHERIT -- refclassid will be RelationRelationId) or * heap_create_with_catalog (CREATE TABLE OF/ALTER TABLE OF -- refclassid will @@ -19557,10 +18782,9 @@ AttachPartitionEnsureIndexes(List **wqueue, Relation rel, Relation attachrel) attachInfos = palloc(sizeof(IndexInfo *) * list_length(attachRelIdxs)); /* Build arrays of all existing indexes and their IndexInfos */ - foreach(cell, attachRelIdxs) + foreach_oid(cldIdxId, attachRelIdxs) { - Oid cldIdxId = lfirst_oid(cell); - int i = foreach_current_index(cell); + int i = foreach_current_index(cldIdxId); attachrelIdxRels[i] = index_open(cldIdxId, AccessShareLock); attachInfos[i] = BuildIndexInfo(attachrelIdxRels[i]); @@ -19694,28 +18918,6 @@ AttachPartitionEnsureIndexes(List **wqueue, Relation rel, Relation attachrel) stmt = generateClonedIndexStmt(NULL, idxRel, attmap, &conOid); - - /* - * If the index is a primary key, mark all columns as NOT NULL if - * they aren't already. - */ - if (stmt->primary) - { - MemoryContextSwitchTo(oldcxt); - for (int j = 0; j < info->ii_NumIndexKeyAttrs; j++) - { - AttrNumber childattno; - - childattno = get_attnum(RelationGetRelid(attachrel), - get_attname(RelationGetRelid(rel), - info->ii_IndexAttrNumbers[j], - false)); - set_attnotnull(wqueue, attachrel, childattno, - true, AccessExclusiveLock); - } - MemoryContextSwitchTo(cxt); - } - DefineIndex(RelationGetRelid(attachrel), stmt, InvalidOid, RelationGetRelid(idxRel), conOid, @@ -20338,7 +19540,7 @@ ATExecDetachPartitionFinalize(Relation rel, RangeVar *name) * DetachAddConstraintIfNeeded * Subroutine for ATExecDetachPartition. Create a constraint that * takes the place of the partition constraint, but avoid creating - * a dupe if a constraint already exists which implies the needed + * a dupe if an constraint already exists which implies the needed * constraint. */ static void @@ -20371,8 +19573,8 @@ DetachAddConstraintIfNeeded(List **wqueue, Relation partRel) n->initially_valid = true; n->skip_validation = true; /* It's a re-add, since it nominally already exists */ - ATAddCheckNNConstraint(wqueue, tab, partRel, n, - true, false, true, ShareUpdateExclusiveLock); + ATAddCheckConstraint(wqueue, tab, partRel, n, + true, false, true, ShareUpdateExclusiveLock); } } @@ -20641,13 +19843,6 @@ ATExecAttachPartitionIdx(List **wqueue, Relation parentIdx, RangeVar *name) RelationGetRelationName(partIdx)))); } - /* - * If it's a primary key, make sure the columns in the partition are - * NOT NULL. - */ - if (parentIdx->rd_index->indisprimary) - verifyPartitionIndexNotNull(childInfo, partTbl); - /* All good -- do it */ IndexSetParentIndex(partIdx, RelationGetRelid(parentIdx)); if (OidIsValid(constraintOid)) @@ -20792,29 +19987,6 @@ validatePartitionedIndex(Relation partedIdx, Relation partedTbl) } /* - * When attaching an index as a partition of a partitioned index which is a - * primary key, verify that all the columns in the partition are marked NOT - * NULL. - */ -static void -verifyPartitionIndexNotNull(IndexInfo *iinfo, Relation partition) -{ - for (int i = 0; i < iinfo->ii_NumIndexKeyAttrs; i++) - { - Form_pg_attribute att = TupleDescAttr(RelationGetDescr(partition), - iinfo->ii_IndexAttrNumbers[i] - 1); - - if (!att->attnotnull) - ereport(ERROR, - errcode(ERRCODE_INVALID_TABLE_DEFINITION), - errmsg("invalid primary key definition"), - errdetail("Column \"%s\" of relation \"%s\" is not marked NOT NULL.", - NameStr(att->attname), - RelationGetRelationName(partition))); - } -} - -/* * Return an OID list of constraints that reference the given relation * that are marked as having a parent constraints. */ diff --git a/src/backend/optimizer/util/plancat.c b/src/backend/optimizer/util/plancat.c index 775c3e26cd8..26f8de77135 100644 --- a/src/backend/optimizer/util/plancat.c +++ b/src/backend/optimizer/util/plancat.c @@ -1680,8 +1680,6 @@ relation_excluded_by_constraints(PlannerInfo *root, * Currently, attnotnull constraints must be treated as NO INHERIT unless * this is a partitioned table. In future we might track their * inheritance status more accurately, allowing this to be refined. - * - * XXX do we need/want to change this? */ include_notnull = (!rte->inh || rte->relkind == RELKIND_PARTITIONED_TABLE); diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y index e8b619926ef..18a0a2dc2b8 100644 --- a/src/backend/parser/gram.y +++ b/src/backend/parser/gram.y @@ -3953,15 +3953,12 @@ ColConstraint: * or be part of a_expr NOT LIKE or similar constructs). */ ColConstraintElem: - NOT NULL_P opt_no_inherit + NOT NULL_P { Constraint *n = makeNode(Constraint); n->contype = CONSTR_NOTNULL; n->location = @1; - n->is_no_inherit = $3; - n->skip_validation = false; - n->initially_valid = true; $$ = (Node *) n; } | NULL_P @@ -4198,20 +4195,6 @@ ConstraintElem: n->initially_valid = !n->skip_validation; $$ = (Node *) n; } - | NOT NULL_P ColId ConstraintAttributeSpec - { - Constraint *n = makeNode(Constraint); - - n->contype = CONSTR_NOTNULL; - n->location = @1; - n->keys = list_make1(makeString($3)); - /* no NOT VALID support yet */ - processCASbits($4, @4, "NOT NULL", - NULL, NULL, NULL, - &n->is_no_inherit, yyscanner); - n->initially_valid = true; - $$ = (Node *) n; - } | UNIQUE opt_unique_null_treatment '(' columnList opt_without_overlaps ')' opt_c_include opt_definition OptConsTableSpace ConstraintAttributeSpec { diff --git a/src/backend/parser/parse_utilcmd.c b/src/backend/parser/parse_utilcmd.c index 0598e897d90..b692d251522 100644 --- a/src/backend/parser/parse_utilcmd.c +++ b/src/backend/parser/parse_utilcmd.c @@ -84,7 +84,6 @@ typedef struct bool isalter; /* true if altering existing table */ List *columns; /* ColumnDef items */ List *ckconstraints; /* CHECK constraints */ - List *nnconstraints; /* NOT NULL constraints */ List *fkconstraints; /* FOREIGN KEY constraints */ List *ixconstraints; /* index-creating constraints */ List *likeclauses; /* LIKE clauses that need post-processing */ @@ -244,7 +243,6 @@ transformCreateStmt(CreateStmt *stmt, const char *queryString) cxt.isalter = false; cxt.columns = NIL; cxt.ckconstraints = NIL; - cxt.nnconstraints = NIL; cxt.fkconstraints = NIL; cxt.ixconstraints = NIL; cxt.likeclauses = NIL; @@ -351,7 +349,6 @@ transformCreateStmt(CreateStmt *stmt, const char *queryString) */ stmt->tableElts = cxt.columns; stmt->constraints = cxt.ckconstraints; - stmt->nnconstraints = cxt.nnconstraints; result = lappend(cxt.blist, stmt); result = list_concat(result, cxt.alist); @@ -550,7 +547,6 @@ transformColumnDefinition(CreateStmtContext *cxt, ColumnDef *column) bool saw_default; bool saw_identity; bool saw_generated; - bool need_notnull = false; ListCell *clist; cxt->columns = lappend(cxt->columns, column); @@ -648,8 +644,10 @@ transformColumnDefinition(CreateStmtContext *cxt, ColumnDef *column) constraint->cooked_expr = NULL; column->constraints = lappend(column->constraints, constraint); - /* have a not-null constraint added later */ - need_notnull = true; + constraint = makeNode(Constraint); + constraint->contype = CONSTR_NOTNULL; + constraint->location = -1; + column->constraints = lappend(column->constraints, constraint); } /* Process column constraints, if any... */ @@ -667,7 +665,7 @@ transformColumnDefinition(CreateStmtContext *cxt, ColumnDef *column) switch (constraint->contype) { case CONSTR_NULL: - if ((saw_nullable && column->is_not_null) || need_notnull) + if (saw_nullable && column->is_not_null) ereport(ERROR, (errcode(ERRCODE_SYNTAX_ERROR), errmsg("conflicting NULL/NOT NULL declarations for column \"%s\" of table \"%s\"", @@ -679,14 +677,6 @@ transformColumnDefinition(CreateStmtContext *cxt, ColumnDef *column) break; case CONSTR_NOTNULL: - if (cxt->ispartitioned && constraint->is_no_inherit) - ereport(ERROR, - errcode(ERRCODE_FEATURE_NOT_SUPPORTED), - errmsg("not-null constraints on partitioned tables cannot be NO INHERIT")); - - /* - * Disallow conflicting [NOT] NULL markings - */ if (saw_nullable && !column->is_not_null) ereport(ERROR, (errcode(ERRCODE_SYNTAX_ERROR), @@ -694,25 +684,8 @@ transformColumnDefinition(CreateStmtContext *cxt, ColumnDef *column) column->colname, cxt->relation->relname), parser_errposition(cxt->pstate, constraint->location))); - /* Ignore redundant NOT NULL markings */ - - /* - * If this is the first time we see this column being marked - * not null, add the constraint entry; and get rid of any - * previous markings to mark the column NOT NULL. - */ - if (!column->is_not_null) - { - column->is_not_null = true; - saw_nullable = true; - - constraint->keys = list_make1(makeString(column->colname)); - cxt->nnconstraints = lappend(cxt->nnconstraints, constraint); - - /* Don't need this anymore, if we had it */ - need_notnull = false; - } - + column->is_not_null = true; + saw_nullable = true; break; case CONSTR_DEFAULT: @@ -762,19 +735,16 @@ transformColumnDefinition(CreateStmtContext *cxt, ColumnDef *column) column->identity = constraint->generated_when; saw_identity = true; - /* - * Identity columns are always NOT NULL, but we may have a - * constraint already. - */ - if (!saw_nullable) - need_notnull = true; - else if (!column->is_not_null) + /* An identity column is implicitly NOT NULL */ + if (saw_nullable && !column->is_not_null) ereport(ERROR, (errcode(ERRCODE_SYNTAX_ERROR), errmsg("conflicting NULL/NOT NULL declarations for column \"%s\" of table \"%s\"", column->colname, cxt->relation->relname), parser_errposition(cxt->pstate, constraint->location))); + column->is_not_null = true; + saw_nullable = true; break; } @@ -881,29 +851,6 @@ transformColumnDefinition(CreateStmtContext *cxt, ColumnDef *column) } /* - * If we need a not-null constraint for SERIAL or IDENTITY, and one was - * not explicitly specified, add one now. - */ - if (need_notnull && !(saw_nullable && column->is_not_null)) - { - Constraint *notnull; - - column->is_not_null = true; - - notnull = makeNode(Constraint); - notnull->contype = CONSTR_NOTNULL; - notnull->conname = NULL; - notnull->deferrable = false; - notnull->initdeferred = false; - notnull->location = -1; - notnull->keys = list_make1(makeString(column->colname)); - notnull->skip_validation = false; - notnull->initially_valid = true; - - cxt->nnconstraints = lappend(cxt->nnconstraints, notnull); - } - - /* * If needed, generate ALTER FOREIGN TABLE ALTER COLUMN statement to add * per-column foreign data wrapper options to this column after creation. */ @@ -972,16 +919,6 @@ transformTableConstraint(CreateStmtContext *cxt, Constraint *constraint) cxt->ckconstraints = lappend(cxt->ckconstraints, constraint); break; - case CONSTR_NOTNULL: - if (cxt->ispartitioned && constraint->is_no_inherit) - ereport(ERROR, - errcode(ERRCODE_FEATURE_NOT_SUPPORTED), - errmsg("not-null constraints on partitioned tables cannot be NO INHERIT")); - - - cxt->nnconstraints = lappend(cxt->nnconstraints, constraint); - break; - case CONSTR_FOREIGN: if (cxt->isforeign) ereport(ERROR, @@ -993,6 +930,7 @@ transformTableConstraint(CreateStmtContext *cxt, Constraint *constraint) break; case CONSTR_NULL: + case CONSTR_NOTNULL: case CONSTR_DEFAULT: case CONSTR_ATTR_DEFERRABLE: case CONSTR_ATTR_NOT_DEFERRABLE: @@ -1028,7 +966,6 @@ transformTableLikeClause(CreateStmtContext *cxt, TableLikeClause *table_like_cla AclResult aclresult; char *comment; ParseCallbackState pcbstate; - bool process_notnull_constraints = false; setup_parser_errposition_callback(&pcbstate, cxt->pstate, table_like_clause->relation->location); @@ -1097,18 +1034,14 @@ transformTableLikeClause(CreateStmtContext *cxt, TableLikeClause *table_like_cla continue; /* - * Create a new column definition + * Create a new column, which is marked as NOT inherited. + * + * For constraints, ONLY the not-null constraint is inherited by the + * new column definition per SQL99. */ def = makeColumnDef(NameStr(attribute->attname), attribute->atttypid, attribute->atttypmod, attribute->attcollation); - - /* - * For constraints, ONLY the not-null constraint is inherited by the - * new column definition per SQL99; however we cannot do that - * correctly here, so we leave it for expandTableLikeClause to handle. - */ - if (attribute->attnotnull) - process_notnull_constraints = true; + def->is_not_null = attribute->attnotnull; /* * Add to column list @@ -1182,78 +1115,20 @@ transformTableLikeClause(CreateStmtContext *cxt, TableLikeClause *table_like_cla * we don't yet know what column numbers the copied columns will have in * the finished table. If any of those options are specified, add the * LIKE clause to cxt->likeclauses so that expandTableLikeClause will be - * called after we do know that; in addition, do that if there are any NOT - * NULL constraints, because those must be propagated even if not - * explicitly requested. - * - * In order for this to work, we remember the relation OID so that + * called after we do know that. Also, remember the relation OID so that * expandTableLikeClause is certain to open the same table. */ - if ((table_like_clause->options & - (CREATE_TABLE_LIKE_DEFAULTS | - CREATE_TABLE_LIKE_GENERATED | - CREATE_TABLE_LIKE_CONSTRAINTS | - CREATE_TABLE_LIKE_INDEXES)) || - process_notnull_constraints) + if (table_like_clause->options & + (CREATE_TABLE_LIKE_DEFAULTS | + CREATE_TABLE_LIKE_GENERATED | + CREATE_TABLE_LIKE_CONSTRAINTS | + CREATE_TABLE_LIKE_INDEXES)) { table_like_clause->relationOid = RelationGetRelid(relation); cxt->likeclauses = lappend(cxt->likeclauses, table_like_clause); } /* - * If INCLUDING INDEXES is not given and a primary key exists, we need to - * add not-null constraints to the columns covered by the PK (except those - * that already have one.) This is required for backwards compatibility. - */ - if ((table_like_clause->options & CREATE_TABLE_LIKE_INDEXES) == 0) - { - Bitmapset *pkcols; - int x = -1; - Bitmapset *donecols = NULL; - ListCell *lc; - - /* - * Obtain a bitmapset of columns on which we'll add not-null - * constraints in expandTableLikeClause, so that we skip this for - * those. - */ - foreach(lc, RelationGetNotNullConstraints(RelationGetRelid(relation), true)) - { - CookedConstraint *cooked = (CookedConstraint *) lfirst(lc); - - donecols = bms_add_member(donecols, cooked->attnum); - } - - pkcols = RelationGetIndexAttrBitmap(relation, - INDEX_ATTR_BITMAP_PRIMARY_KEY); - while ((x = bms_next_member(pkcols, x)) >= 0) - { - Constraint *notnull; - AttrNumber attnum = x + FirstLowInvalidHeapAttributeNumber; - Form_pg_attribute attForm; - - /* ignore if we already have one for this column */ - if (bms_is_member(attnum, donecols)) - continue; - - attForm = TupleDescAttr(tupleDesc, attnum - 1); - - notnull = makeNode(Constraint); - notnull->contype = CONSTR_NOTNULL; - notnull->conname = NULL; - notnull->is_no_inherit = false; - notnull->deferrable = false; - notnull->initdeferred = false; - notnull->location = -1; - notnull->keys = list_make1(makeString(pstrdup(NameStr(attForm->attname)))); - notnull->skip_validation = false; - notnull->initially_valid = true; - - cxt->nnconstraints = lappend(cxt->nnconstraints, notnull); - } - } - - /* * We may copy extended statistics if requested, since the representation * of CreateStatsStmt doesn't depend on column numbers. */ @@ -1319,8 +1194,6 @@ expandTableLikeClause(RangeVar *heapRel, TableLikeClause *table_like_clause) TupleConstr *constr; AttrMap *attmap; char *comment; - bool at_pushed = false; - ListCell *lc; /* * Open the relation referenced by the LIKE clause. We should still have @@ -1492,20 +1365,6 @@ expandTableLikeClause(RangeVar *heapRel, TableLikeClause *table_like_clause) } /* - * Copy not-null constraints, too (these do not require any option to have - * been given). - */ - foreach(lc, RelationGetNotNullConstraints(RelationGetRelid(relation), false)) - { - AlterTableCmd *atsubcmd; - - atsubcmd = makeNode(AlterTableCmd); - atsubcmd->subtype = AT_AddConstraint; - atsubcmd->def = (Node *) lfirst_node(Constraint, lc); - atsubcmds = lappend(atsubcmds, atsubcmd); - } - - /* * If we generated any ALTER TABLE actions above, wrap them into a single * ALTER TABLE command. Stick it at the front of the result, so it runs * before any CommentStmts we made above. @@ -1519,8 +1378,6 @@ expandTableLikeClause(RangeVar *heapRel, TableLikeClause *table_like_clause) atcmd->objtype = OBJECT_TABLE; atcmd->missing_ok = false; result = lcons(atcmd, result); - - at_pushed = true; } /* @@ -1548,39 +1405,6 @@ expandTableLikeClause(RangeVar *heapRel, TableLikeClause *table_like_clause) attmap, NULL); - /* - * The PK columns might not yet non-nullable, so make sure they - * become so. - */ - if (index_stmt->primary) - { - foreach(lc, index_stmt->indexParams) - { - IndexElem *col = lfirst_node(IndexElem, lc); - AlterTableCmd *notnullcmd = makeNode(AlterTableCmd); - - notnullcmd->subtype = AT_SetAttNotNull; - notnullcmd->name = pstrdup(col->name); - /* Luckily we can still add more AT-subcmds here */ - atsubcmds = lappend(atsubcmds, notnullcmd); - } - - /* - * If we had already put the AlterTableStmt into the output - * list, we don't need to do so again; otherwise do it. - */ - if (!at_pushed) - { - AlterTableStmt *atcmd = makeNode(AlterTableStmt); - - atcmd->relation = copyObject(heapRel); - atcmd->cmds = atsubcmds; - atcmd->objtype = OBJECT_TABLE; - atcmd->missing_ok = false; - result = lcons(atcmd, result); - } - } - /* Copy comment on index, if requested */ if (table_like_clause->options & CREATE_TABLE_LIKE_COMMENTS) { @@ -1661,8 +1485,8 @@ transformOfType(CreateStmtContext *cxt, TypeName *ofTypename) * with the index there. * * Unlike transformIndexConstraint, we don't make any effort to force primary - * key columns to be not-null. The larger cloning process this is part of - * should have cloned their not-null status separately (and DefineIndex will + * key columns to be NOT NULL. The larger cloning process this is part of + * should have cloned their NOT NULL status separately (and DefineIndex will * complain if that fails to happen). */ IndexStmt * @@ -2210,12 +2034,10 @@ transformIndexConstraints(CreateStmtContext *cxt) ListCell *lc; /* - * Run through the constraints that need to generate an index, and do so. - * - * For PRIMARY KEY, in addition we set each column's attnotnull flag true. - * We do not create a separate not-null constraint, as that would be - * redundant: the PRIMARY KEY constraint itself fulfills that role. Other - * constraint types don't need any not-null markings. + * Run through the constraints that need to generate an index. For PRIMARY + * KEY, mark each column as NOT NULL and create an index. For UNIQUE or + * EXCLUDE, create an index as for PRIMARY KEY, but do not insist on NOT + * NULL. */ foreach(lc, cxt->ixconstraints) { @@ -2289,7 +2111,9 @@ transformIndexConstraints(CreateStmtContext *cxt) } /* - * Now append all the IndexStmts to cxt->alist. + * Now append all the IndexStmts to cxt->alist. If we generated an ALTER + * TABLE SET NOT NULL statement to support a primary key, it's already in + * cxt->alist. */ cxt->alist = list_concat(cxt->alist, finalindexlist); } @@ -2297,10 +2121,12 @@ transformIndexConstraints(CreateStmtContext *cxt) /* * transformIndexConstraint * Transform one UNIQUE, PRIMARY KEY, or EXCLUDE constraint for - * transformIndexConstraints. An IndexStmt is returned. + * transformIndexConstraints. * - * For a PRIMARY KEY constraint, we additionally force the columns to be - * marked as not-null, without producing a not-null constraint. + * We return an IndexStmt. For a PRIMARY KEY constraint, we additionally + * produce not-null constraints, either by marking ColumnDefs in cxt->columns + * as is_not_null or by adding an ALTER TABLE SET NOT NULL command to + * cxt->alist. */ static IndexStmt * transformIndexConstraint(Constraint *constraint, CreateStmtContext *cxt) @@ -2564,7 +2390,7 @@ transformIndexConstraint(Constraint *constraint, CreateStmtContext *cxt) * For UNIQUE and PRIMARY KEY, we just have a list of column names. * * Make sure referenced keys exist. If we are making a PRIMARY KEY index, - * also make sure they are not-null. + * also make sure they are NOT NULL. */ else { @@ -2572,6 +2398,7 @@ transformIndexConstraint(Constraint *constraint, CreateStmtContext *cxt) { char *key = strVal(lfirst(lc)); bool found = false; + bool forced_not_null = false; ColumnDef *column = NULL; ListCell *columns; IndexElem *iparam; @@ -2592,14 +2419,13 @@ transformIndexConstraint(Constraint *constraint, CreateStmtContext *cxt) * column is defined in the new table. For PRIMARY KEY, we * can apply the not-null constraint cheaply here ... unless * the column is marked is_from_type, in which case marking it - * here would be ineffective (see MergeAttributes). Note that - * this isn't effective in ALTER TABLE either, unless the - * column is being added in the same command. + * here would be ineffective (see MergeAttributes). */ if (constraint->contype == CONSTR_PRIMARY && !column->is_from_type) { column->is_not_null = true; + forced_not_null = true; } } else if (SystemAttributeByName(key) != NULL) @@ -2607,7 +2433,7 @@ transformIndexConstraint(Constraint *constraint, CreateStmtContext *cxt) /* * column will be a system column in the new table, so accept * it. System columns can't ever be null, so no need to worry - * about PRIMARY/NOT NULL constraint. + * about PRIMARY/not-null constraint. */ found = true; } @@ -2642,6 +2468,14 @@ transformIndexConstraint(Constraint *constraint, CreateStmtContext *cxt) if (strcmp(key, inhname) == 0) { found = true; + + /* + * It's tempting to set forced_not_null if the + * parent column is already NOT NULL, but that + * seems unsafe because the column's NOT NULL + * marking might disappear between now and + * execution. Do the runtime check to be safe. + */ break; } } @@ -2695,11 +2529,15 @@ transformIndexConstraint(Constraint *constraint, CreateStmtContext *cxt) iparam->nulls_ordering = SORTBY_NULLS_DEFAULT; index->indexParams = lappend(index->indexParams, iparam); - if (constraint->contype == CONSTR_PRIMARY) + /* + * For a primary-key column, also create an item for ALTER TABLE + * SET NOT NULL if we couldn't ensure it via is_not_null above. + */ + if (constraint->contype == CONSTR_PRIMARY && !forced_not_null) { AlterTableCmd *notnullcmd = makeNode(AlterTableCmd); - notnullcmd->subtype = AT_SetAttNotNull; + notnullcmd->subtype = AT_SetNotNull; notnullcmd->name = pstrdup(key); notnullcmds = lappend(notnullcmds, notnullcmd); } @@ -3642,7 +3480,6 @@ transformAlterTableStmt(Oid relid, AlterTableStmt *stmt, cxt.isalter = true; cxt.columns = NIL; cxt.ckconstraints = NIL; - cxt.nnconstraints = NIL; cxt.fkconstraints = NIL; cxt.ixconstraints = NIL; cxt.likeclauses = NIL; @@ -3912,8 +3749,8 @@ transformAlterTableStmt(Oid relid, AlterTableStmt *stmt, /* * We assume here that cxt.alist contains only IndexStmts and possibly - * AT_SetAttNotNull statements generated from primary key constraints. - * We absorb the subcommands of the latter directly. + * ALTER TABLE SET NOT NULL statements generated from primary key + * constraints. We absorb the subcommands of the latter directly. */ if (IsA(istmt, IndexStmt)) { @@ -3936,7 +3773,7 @@ transformAlterTableStmt(Oid relid, AlterTableStmt *stmt, } cxt.alist = NIL; - /* Append any CHECK, NOT NULL or FK constraints to the commands list */ + /* Append any CHECK or FK constraints to the commands list */ foreach(l, cxt.ckconstraints) { newcmd = makeNode(AlterTableCmd); @@ -3944,13 +3781,6 @@ transformAlterTableStmt(Oid relid, AlterTableStmt *stmt, newcmd->def = (Node *) lfirst_node(Constraint, l); newcmds = lappend(newcmds, newcmd); } - foreach(l, cxt.nnconstraints) - { - newcmd = makeNode(AlterTableCmd); - newcmd->subtype = AT_AddConstraint; - newcmd->def = (Node *) lfirst_node(Constraint, l); - newcmds = lappend(newcmds, newcmd); - } foreach(l, cxt.fkconstraints) { newcmd = makeNode(AlterTableCmd); diff --git a/src/backend/utils/adt/ruleutils.c b/src/backend/utils/adt/ruleutils.c index 302cd8e7f33..9a6d372414c 100644 --- a/src/backend/utils/adt/ruleutils.c +++ b/src/backend/utils/adt/ruleutils.c @@ -2506,28 +2506,6 @@ pg_get_constraintdef_worker(Oid constraintId, bool fullCommand, conForm->connoinherit ? " NO INHERIT" : ""); break; } - case CONSTRAINT_NOTNULL: - { - if (conForm->conrelid) - { - AttrNumber attnum; - - attnum = extractNotNullColumn(tup); - - appendStringInfo(&buf, "NOT NULL %s", - quote_identifier(get_attname(conForm->conrelid, - attnum, false))); - if (((Form_pg_constraint) GETSTRUCT(tup))->connoinherit) - appendStringInfoString(&buf, " NO INHERIT"); - } - else if (conForm->contypid) - { - /* conkey is null for domain not-null constraints */ - appendStringInfoString(&buf, "NOT NULL"); - } - break; - } - case CONSTRAINT_TRIGGER: /* diff --git a/src/backend/utils/cache/relcache.c b/src/backend/utils/cache/relcache.c index 262c9878dd3..e6072cbdd9e 100644 --- a/src/backend/utils/cache/relcache.c +++ b/src/backend/utils/cache/relcache.c @@ -4810,46 +4810,18 @@ RelationGetIndexList(Relation relation) result = lappend_oid(result, index->indexrelid); /* - * Non-unique or predicate indexes aren't interesting for either oid - * indexes or replication identity indexes, so don't check them. - * Deferred ones are not useful for replication identity either; but - * we do include them if they are PKs. + * Invalid, non-unique, non-immediate or predicate indexes aren't + * interesting for either oid indexes or replication identity indexes, + * so don't check them. */ - if (!index->indisunique || + if (!index->indisvalid || !index->indisunique || + !index->indimmediate || !heap_attisnull(htup, Anum_pg_index_indpred, NULL)) continue; - /* - * Remember primary key index, if any. We do this only if the index - * is valid; but if the table is partitioned, then we do it even if - * it's invalid. - * - * The reason for returning invalid primary keys for foreign tables is - * because of pg_dump of NOT NULL constraints, and the fact that PKs - * remain marked invalid until the partitions' PKs are attached to it. - * If we make rd_pkindex invalid, then the attnotnull flag is reset - * after the PK is created, which causes the ALTER INDEX ATTACH - * PARTITION to fail with 'column ... is not marked NOT NULL'. With - * this, dropconstraint_internal() will believe that the columns must - * not have attnotnull reset, so the PKs-on-partitions can be attached - * correctly, until finally the PK-on-parent is marked valid. - * - * Also, this doesn't harm anything, because rd_pkindex is not a - * "real" index anyway, but a RELKIND_PARTITIONED_INDEX. - */ - if (index->indisprimary && - (index->indisvalid || - relation->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)) - { + /* remember primary key index if any */ + if (index->indisprimary) pkeyIndex = index->indexrelid; - pkdeferrable = !index->indimmediate; - } - - if (!index->indimmediate) - continue; - - if (!index->indisvalid) - continue; /* remember explicitly chosen replica index */ if (index->indisreplident) |