diff options
Diffstat (limited to 'src/backend')
| -rw-r--r-- | src/backend/catalog/heap.c | 491 | ||||
| -rw-r--r-- | src/backend/catalog/pg_constraint.c | 97 | ||||
| -rw-r--r-- | src/backend/commands/tablecmds.c | 1326 | ||||
| -rw-r--r-- | src/backend/nodes/outfuncs.c | 4 | ||||
| -rw-r--r-- | src/backend/nodes/readfuncs.c | 8 | ||||
| -rw-r--r-- | src/backend/optimizer/util/plancat.c | 2 | ||||
| -rw-r--r-- | src/backend/parser/gram.y | 13 | ||||
| -rw-r--r-- | src/backend/parser/parse_utilcmd.c | 206 | ||||
| -rw-r--r-- | src/backend/utils/adt/ruleutils.c | 14 |
9 files changed, 1677 insertions, 484 deletions
diff --git a/src/backend/catalog/heap.c b/src/backend/catalog/heap.c index 2a0d82aedd7..d9b1adfe12a 100644 --- a/src/backend/catalog/heap.c +++ b/src/backend/catalog/heap.c @@ -2161,6 +2161,57 @@ 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) +{ + int16 attNos; + Oid constrOid; + + /* We only ever store one column per constraint */ + attNos = attnum; + + constrOid = + CreateConstraintEntry(nnname, + RelationGetNamespace(rel), + CONSTRAINT_NOTNULL, + false, + false, + is_validated, + InvalidOid, + RelationGetRelid(rel), + &attNos, + 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); + return constrOid; +} + +/* * Store defaults and constraints (passed as a list of CookedConstraint). * * Each CookedConstraint struct is modified to store the new catalog tuple OID. @@ -2204,6 +2255,14 @@ 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); @@ -2259,6 +2318,7 @@ AddRelationNewConstraints(Relation rel, ParseNamespaceItem *nsitem; int numchecks; List *checknames; + List *nnnames; ListCell *cell; Node *expr; CookedConstraint *cooked; @@ -2344,130 +2404,179 @@ AddRelationNewConstraints(Relation rel, */ numchecks = numoldchecks; checknames = NIL; + nnnames = NIL; foreach(cell, newConstraints) { Constraint *cdef = (Constraint *) lfirst(cell); - char *ccname; Oid constrOid; - if (cdef->contype != CONSTR_CHECK) - continue; - - if (cdef->raw_expr != NULL) + if (cdef->contype == CONSTR_CHECK) { - Assert(cdef->cooked_expr == NULL); + char *ccname; /* - * Transform raw parsetree to executable expression, and verify - * it's valid as a CHECK constraint. + * XXX Should we detect the case with CHECK (foo IS NOT NULL) and + * handle it as a NOT NULL constraint? */ - expr = cookConstraint(pstate, cdef->raw_expr, - RelationGetRelationName(rel)); - } - else - { - Assert(cdef->cooked_expr != NULL); + + if (cdef->raw_expr != NULL) + { + Assert(cdef->cooked_expr == NULL); + + /* + * Transform raw parsetree to executable expression, and + * verify it's valid as a CHECK constraint. + */ + expr = cookConstraint(pstate, cdef->raw_expr, + RelationGetRelationName(rel)); + } + else + { + Assert(cdef->cooked_expr != NULL); + + /* + * Here, we assume the parser will only pass us valid CHECK + * expressions, so we do no particular checking. + */ + expr = stringToNode(cdef->cooked_expr); + } /* - * Here, we assume the parser will only pass us valid CHECK - * expressions, so we do no particular checking. + * Check name uniqueness, or generate a name if none was given. */ - expr = stringToNode(cdef->cooked_expr); - } + if (cdef->conname != NULL) + { + ListCell *cell2; - /* - * Check name uniqueness, or generate a name if none was given. - */ - 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) + { + if (strcmp((char *) lfirst(cell2), ccname) == 0) + ereport(ERROR, + (errcode(ERRCODE_DUPLICATE_OBJECT), + errmsg("check constraint \"%s\" already exists", + ccname))); + } - ccname = cdef->conname; - /* Check against other new constraints */ - /* Needed because we don't do CommandCounterIncrement in loop */ - foreach(cell2, checknames) - { - if (strcmp((char *) lfirst(cell2), ccname) == 0) - ereport(ERROR, - (errcode(ERRCODE_DUPLICATE_OBJECT), - errmsg("check constraint \"%s\" already exists", - ccname))); + /* save name for future checks */ + checknames = lappend(checknames, ccname); + + /* + * Check against pre-existing constraints. If we are allowed + * to merge with an existing constraint, there's no more to do + * here. (We omit the duplicate constraint from the result, + * which is what ATAddCheckConstraint wants.) + */ + if (MergeWithExistingConstraint(rel, ccname, expr, + allow_merge, is_local, + cdef->initially_valid, + cdef->is_no_inherit)) + continue; } + else + { + /* + * When generating a name, we want to create "tab_col_check" + * for a column constraint and "tab_check" for a table + * constraint. We no longer have any info about the syntactic + * positioning of the constraint phrase, so we approximate + * this by seeing whether the expression references more than + * one column. (If the user played by the rules, the result + * is the same...) + * + * Note: pull_var_clause() doesn't descend into sublinks, but + * we eliminated those above; and anyway this only needs to be + * an approximate answer. + */ + List *vars; + char *colname; + + vars = pull_var_clause(expr, 0); + + /* eliminate duplicates */ + vars = list_union(NIL, vars); + + if (list_length(vars) == 1) + colname = get_attname(RelationGetRelid(rel), + ((Var *) linitial(vars))->varattno, + true); + else + colname = NULL; - /* save name for future checks */ - checknames = lappend(checknames, ccname); + ccname = ChooseConstraintName(RelationGetRelationName(rel), + colname, + "check", + RelationGetNamespace(rel), + checknames); + + /* save name for future checks */ + checknames = lappend(checknames, ccname); + } /* - * Check against pre-existing constraints. If we are allowed to - * merge with an existing constraint, there's no more to do here. - * (We omit the duplicate constraint from the result, which is - * what ATAddCheckConstraint wants.) + * OK, store it. */ - if (MergeWithExistingConstraint(rel, ccname, expr, - allow_merge, is_local, - cdef->initially_valid, - cdef->is_no_inherit)) - continue; + constrOid = + StoreRelCheck(rel, ccname, expr, cdef->initially_valid, is_local, + is_local ? 0 : 1, cdef->is_no_inherit, is_internal); + + numchecks++; + + cooked = (CookedConstraint *) palloc(sizeof(CookedConstraint)); + cooked->contype = CONSTR_CHECK; + cooked->conoid = constrOid; + cooked->name = ccname; + cooked->attnum = 0; + cooked->expr = expr; + cooked->skip_validation = cdef->skip_validation; + cooked->is_local = is_local; + cooked->inhcount = is_local ? 0 : 1; + cooked->is_no_inherit = cdef->is_no_inherit; + cookedConstraints = lappend(cookedConstraints, cooked); } - else + else if (cdef->contype == CONSTR_NOTNULL) { - /* - * When generating a name, we want to create "tab_col_check" for a - * column constraint and "tab_check" for a table constraint. We - * no longer have any info about the syntactic positioning of the - * constraint phrase, so we approximate this by seeing whether the - * expression references more than one column. (If the user - * played by the rules, the result is the same...) - * - * Note: pull_var_clause() doesn't descend into sublinks, but we - * eliminated those above; and anyway this only needs to be an - * approximate answer. - */ - List *vars; - char *colname; - - vars = pull_var_clause(expr, 0); + CookedConstraint *nncooked; + AttrNumber colnum; + char *nnname; - /* eliminate duplicates */ - vars = list_union(NIL, vars); + colnum = get_attnum(RelationGetRelid(rel), + cdef->colname); + if (colnum == InvalidAttrNumber) + elog(ERROR, "invalid column name \"%s\"", cdef->colname); - if (list_length(vars) == 1) - colname = get_attname(RelationGetRelid(rel), - ((Var *) linitial(vars))->varattno, - true); + if (cdef->conname) + nnname = cdef->conname; /* verify clash? */ else - colname = NULL; - - ccname = ChooseConstraintName(RelationGetRelationName(rel), - colname, - "check", - RelationGetNamespace(rel), - checknames); - - /* save name for future checks */ - checknames = lappend(checknames, ccname); + nnname = ChooseConstraintName(RelationGetRelationName(rel), + cdef->colname, + "not_null", + RelationGetNamespace(rel), + nnnames); + nnnames = lappend(nnnames, nnname); + + constrOid = + StoreRelNotNull(rel, nnname, colnum, + cdef->initially_valid, + is_local, + is_local ? 0 : 1, + 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 = is_local ? 0 : 1; + nncooked->is_no_inherit = cdef->is_no_inherit; + + cookedConstraints = lappend(cookedConstraints, nncooked); } - - /* - * OK, store it. - */ - constrOid = - StoreRelCheck(rel, ccname, expr, cdef->initially_valid, is_local, - is_local ? 0 : 1, cdef->is_no_inherit, is_internal); - - numchecks++; - - cooked = (CookedConstraint *) palloc(sizeof(CookedConstraint)); - cooked->contype = CONSTR_CHECK; - cooked->conoid = constrOid; - cooked->name = ccname; - cooked->attnum = 0; - cooked->expr = expr; - cooked->skip_validation = cdef->skip_validation; - cooked->is_local = is_local; - cooked->inhcount = is_local ? 0 : 1; - cooked->is_no_inherit = cdef->is_no_inherit; - cookedConstraints = lappend(cookedConstraints, cooked); } /* @@ -2637,6 +2746,190 @@ 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; + + if (v1 < v2) + return -1; + if (v1 > v2) + return 1; + return 0; +} + +/* + * Create the NOT NULL constraints for the 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 *nnnames = NIL; + List *givennames = NIL; + List *nncols = NIL; + ListCell *lc; + + /* + * First, create all NOT NULLs 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; + + attnum = get_attnum(RelationGetRelid(rel), constr->colname); + + foreach(lc2, old_notnulls) + { + CookedConstraint *old = (CookedConstraint *) lfirst(lc2); + + if (old->attnum == attnum) + { + 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), conname) == 0) + ereport(ERROR, + errmsg("constraint name \"%s\" is already in use in relation \"%s\"", + 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 additional_notnulls list, we must create a + * NOT NULL constraint marked not-local. Because multiple parents could + * specify a NOT NULL for the same column, we must count how many there + * are and set inhcount accordingly, deleting elements we've already + * processed. + * + * We don't use foreach() here because we have two nested loops over the + * cooked 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 inhcount = 1; + ListCell *lc2; + + cooked = (CookedConstraint *) list_nth(old_notnulls, outerpos); + + /* We just preserve the first 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; + + 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, + false, 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/pg_constraint.c b/src/backend/catalog/pg_constraint.c index 4002317f705..df73678cced 100644 --- a/src/backend/catalog/pg_constraint.c +++ b/src/backend/catalog/pg_constraint.c @@ -563,6 +563,103 @@ ChooseConstraintName(const char *name1, const char *name2, } /* + * Find and return 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(Relation rel, 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(RelationGetRelid(rel))); + 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(Relation rel, const char *colname) +{ + AttrNumber attnum = get_attnum(RelationGetRelid(rel), colname); + + return findNotNullConstraintAttnum(rel, attnum); +} + +/* + * 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 CHECK 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; +} + +/* * Delete a single constraint record. */ void diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c index d9bbeafd82c..03466e495ac 100644 --- a/src/backend/commands/tablecmds.c +++ b/src/backend/commands/tablecmds.c @@ -203,7 +203,7 @@ typedef struct AlteredTableInfo typedef struct NewConstraint { char *name; /* Constraint name, or NULL if none */ - ConstrType contype; /* CHECK or FOREIGN */ + ConstrType contype; /* CHECK, FOREIGN */ Oid refrelid; /* PK rel, if FOREIGN */ Oid refindid; /* OID of PK's index, if FOREIGN */ Oid conid; /* OID of pg_constraint entry, if FOREIGN */ @@ -350,7 +350,8 @@ static void truncate_check_activity(Relation rel); static void RangeVarCallbackForTruncate(const RangeVar *relation, Oid relId, Oid oldRelId, void *arg); static List *MergeAttributes(List *schema, List *supers, char relpersistence, - bool is_partition, List **supconstr); + bool is_partition, List **supconstr, + List **additional_notnulls); static bool MergeCheckConstraint(List *constraints, char *name, Node *expr); static void MergeAttributesIntoExisting(Relation child_rel, Relation parent_rel); static void MergeConstraintsIntoExisting(Relation child_rel, Relation parent_rel); @@ -431,14 +432,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 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 ObjectAddress ATExecDropNotNull(Relation rel, const char *colName, bool recurse, + LOCKMODE lockmode); +static void 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 void ATExecSetAttNotNull(List **wqueue, 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); @@ -541,6 +544,11 @@ static void ATExecDropConstraint(Relation rel, const char *constrName, 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, @@ -616,7 +624,7 @@ static void RemoveInheritance(Relation child_rel, Relation parent_rel, static ObjectAddress ATExecAttachPartition(List **wqueue, Relation rel, PartitionCmd *cmd, AlterTableUtilityContext *context); -static void AttachPartitionEnsureIndexes(Relation rel, Relation attachrel); +static void AttachPartitionEnsureIndexes(List **wqueue, Relation rel, Relation attachrel); static void QueuePartitionConstraintValidation(List **wqueue, Relation scanrel, List *partConstraint, bool validate_default); @@ -634,6 +642,7 @@ 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, char *compression); @@ -671,8 +680,10 @@ 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; @@ -862,12 +873,13 @@ DefineRelation(CreateStmt *stmt, char relkind, Oid ownerId, MergeAttributes(stmt->tableElts, inheritOids, stmt->relation->relpersistence, stmt->partbound != NULL, - &old_constraints); + &old_constraints, &old_notnulls); /* * Create a tuple descriptor from the relation schema. Note that this - * deals with column names, types, and NOT NULL constraints, but not - * default values or CHECK constraints; we handle those below. + * deals with column names, types, and in-descriptor NOT NULL flags, but + * not default values, NOT NULL or CHECK constraints; we handle those + * below. */ descriptor = BuildDescForRelation(stmt->tableElts); @@ -1250,6 +1262,17 @@ DefineRelation(CreateStmt *stmt, char relkind, Oid ownerId, AddRelationNewConstraints(rel, NIL, stmt->constraints, true, true, false, queryString); + /* + * Finally, merge the NOT NULL constraints that are directly declared 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); /* @@ -2298,6 +2321,8 @@ 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. + * 'nnconstraints' receives a list of CookedConstraints that corresponds to + * constraints coming from inheritance parents. * * Return value: * Completed schema list. @@ -2328,7 +2353,10 @@ 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. + * 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.) * * The default value for a child column is defined as: * (1) If the child schema specifies a default, that value is used. @@ -2347,10 +2375,11 @@ storage_name(char c) */ static List * MergeAttributes(List *schema, List *supers, char relpersistence, - bool is_partition, List **supconstr) + bool is_partition, List **supconstr, List **supnotnulls) { List *inhSchema = NIL; List *constraints = NIL; + List *nnconstraints = NIL; bool have_bogus_defaults = false; int child_attno; static Node bogus_marker = {0}; /* marks conflicting defaults */ @@ -2462,9 +2491,13 @@ MergeAttributes(List *schema, List *supers, char relpersistence, AttrMap *newattmap; List *inherited_defaults; List *cols_with_defaults; + List *nnconstrs; AttrNumber parent_attno; ListCell *lc1; ListCell *lc2; + Bitmapset *pkattrs; + Bitmapset *nncols = NULL; + /* caller already got lock */ relation = table_open(parent, NoLock); @@ -2553,6 +2586,20 @@ MergeAttributes(List *schema, 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(relation, true); + foreach(lc1, nnconstrs) + nncols = bms_add_member(nncols, + ((CookedConstraint *) lfirst(lc1))->attnum); + for (parent_attno = 1; parent_attno <= tupleDesc->natts; parent_attno++) { @@ -2648,9 +2695,38 @@ MergeAttributes(List *schema, List *supers, char relpersistence, } /* - * Merge of NOT NULL constraints = OR 'em together + * In regular inheritance, columns in the parent's primary key + * get an extra CHECK (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 = exist_attno; + nn->expr = NULL; + nn->skip_validation = false; + nn->is_local = false; + nn->inhcount = 1; + nn->is_no_inherit = false; + + nnconstraints = lappend(nnconstraints, nn); + } + + /* + * mark attnotnull if parent has it and it's not NO INHERIT */ - def->is_not_null |= attribute->attnotnull; + if (bms_is_member(parent_attno, nncols) || + bms_is_member(parent_attno - FirstLowInvalidHeapAttributeNumber, + pkattrs)) + def->is_not_null = true; /* * Check for GENERATED conflicts @@ -2684,7 +2760,11 @@ MergeAttributes(List *schema, List *supers, char relpersistence, attribute->atttypmod); def->inhcount = 1; def->is_local = false; - def->is_not_null = attribute->attnotnull; + /* 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)) + def->is_not_null = true; def->is_from_type = false; def->storage = attribute->attstorage; def->raw_default = NULL; @@ -2701,6 +2781,33 @@ MergeAttributes(List *schema, List *supers, char relpersistence, def->compression = NULL; inhSchema = lappend(inhSchema, def); newattmap->attnums[parent_attno - 1] = ++child_attno; + + /* + * 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); + } } /* @@ -2845,6 +2952,19 @@ MergeAttributes(List *schema, 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); + + nn->attnum = newattmap->attnums[nn->attnum - 1]; + + nnconstraints = lappend(nnconstraints, nn); + } + free_attrmap(newattmap); /* @@ -3051,8 +3171,7 @@ MergeAttributes(List *schema, 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, we merge parent's NOT NULL constraints and - * defaults into each corresponding column definition. + * actually exist. */ if (is_partition) { @@ -3158,6 +3277,8 @@ MergeAttributes(List *schema, List *supers, char relpersistence, } *supconstr = constraints; + *supnotnulls = nnconstraints; + return schema; } @@ -3209,6 +3330,85 @@ MergeCheckConstraint(List *constraints, char *name, Node *expr) return false; } +/* + * RelationGetNotNullConstraints -- get list of NOT NULL constraints + * + * 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(Relation relation, 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(RelationGetRelid(relation))); + 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->colname = get_attname(RelationGetRelid(relation), + colnum, false); + constr->skip_validation = false; + constr->initially_valid = true; + notnulls = lappend(notnulls, constr); + } + } + + systable_endscan(conscan); + table_close(constrRel, AccessShareLock); + + return notnulls; +} /* * StoreCatalogInheritance @@ -3769,7 +3969,10 @@ rename_constraint_internal(Oid myrelid, constraintOid); con = (Form_pg_constraint) GETSTRUCT(tuple); - if (myrelid && con->contype == CONSTRAINT_CHECK && !con->connoinherit) + if (myrelid && + (con->contype == CONSTRAINT_CHECK || + con->contype == CONSTRAINT_NOTNULL) && + !con->connoinherit) { if (recurse) { @@ -4354,6 +4557,7 @@ AlterTableGetLockLevel(List *cmds) case AT_AddIndexConstraint: case AT_ReplicaIdentity: case AT_SetNotNull: + case AT_SetAttNotNull: case AT_EnableRowSecurity: case AT_DisableRowSecurity: case AT_ForceRowSecurity: @@ -4652,15 +4856,23 @@ 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); - ATPrepDropNotNull(rel, recurse, recursing); - ATSimpleRecursion(wqueue, rel, cmd, recurse, lockmode, context); + /* Set up recursion for phase 2; no other prep needed */ + if (recurse) + cmd->recurse = true; pass = AT_PASS_DROP; break; case AT_SetNotNull: /* ALTER COLUMN SET NOT NULL */ ATSimplePermissions(cmd->subtype, rel, ATT_TABLE | ATT_FOREIGN_TABLE); /* Need command-specific recursion decision */ - ATPrepSetNotNull(wqueue, rel, cmd, recurse, recursing, - lockmode, context); + if (recurse) + cmd->recurse = true; + pass = AT_PASS_COL_ATTRS; + break; + case AT_SetAttNotNull: /* set pg_attribute.attnotnull without adding + * a constraint */ + ATSimplePermissions(cmd->subtype, rel, ATT_TABLE | ATT_FOREIGN_TABLE); + /* Need command-specific recursion decision */ + ATSimpleRecursion(wqueue, rel, cmd, recurse, lockmode, context); pass = AT_PASS_COL_ATTRS; break; case AT_CheckNotNull: /* check column is already marked NOT NULL */ @@ -5045,10 +5257,14 @@ ATExecCmd(List **wqueue, AlteredTableInfo *tab, address = ATExecDropIdentity(rel, cmd->name, cmd->missing_ok, lockmode); break; case AT_DropNotNull: /* ALTER COLUMN DROP NOT NULL */ - address = ATExecDropNotNull(rel, cmd->name, lockmode); + address = ATExecDropNotNull(rel, cmd->name, cmd->recurse, lockmode); break; case AT_SetNotNull: /* ALTER COLUMN SET NOT NULL */ - address = ATExecSetNotNull(tab, rel, cmd->name, lockmode); + address = ATExecSetNotNull(wqueue, rel, NULL, cmd->name, + cmd->recurse, false, NULL, lockmode); + break; + case AT_SetAttNotNull: /* set pg_attribute.attnotnull */ + ATExecSetAttNotNull(wqueue, rel, cmd->name, lockmode); break; case AT_CheckNotNull: /* check column is already marked NOT NULL */ ATExecCheckNotNull(tab, rel, cmd->name, lockmode); @@ -5387,11 +5603,8 @@ ATParseTransformCmd(List **wqueue, AlteredTableInfo *tab, Relation rel, */ switch (cmd2->subtype) { - case AT_SetNotNull: - /* Need command-specific recursion decision */ - ATPrepSetNotNull(wqueue, rel, cmd2, - recurse, false, - lockmode, context); + case AT_SetAttNotNull: + ATSimpleRecursion(wqueue, rel, cmd2, recurse, lockmode, context); pass = AT_PASS_COL_ATTRS; break; case AT_AddIndex: @@ -6067,6 +6280,7 @@ 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; @@ -6175,6 +6389,8 @@ 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_DropExpression: return "ALTER COLUMN ... DROP EXPRESSION"; case AT_CheckNotNull: @@ -6774,8 +6990,7 @@ ATPrepAddColumn(List **wqueue, Relation rel, bool recurse, bool recursing, */ static ObjectAddress ATExecAddColumn(List **wqueue, AlteredTableInfo *tab, Relation rel, - AlterTableCmd **cmd, - bool recurse, bool recursing, + AlterTableCmd **cmd, bool recurse, bool recursing, LOCKMODE lockmode, int cur_pass, AlterTableUtilityContext *context) { @@ -7290,41 +7505,19 @@ 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, LOCKMODE lockmode) +ATExecDropNotNull(Relation rel, const char *colName, bool recurse, + LOCKMODE lockmode) { HeapTuple tuple; + HeapTuple conTup; Form_pg_attribute attTup; AttrNumber attnum; Relation attr_rel; - List *indexoidlist; - ListCell *indexoidscan; ObjectAddress address; /* @@ -7340,6 +7533,15 @@ ATExecDropNotNull(Relation rel, const char *colName, LOCKMODE lockmode) 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) @@ -7355,68 +7557,43 @@ ATExecDropNotNull(Relation rel, const char *colName, LOCKMODE lockmode) colName, RelationGetRelationName(rel)))); /* - * 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. + * It's not OK to remove a constraint only for the parent and leave it in + * the children, so disallow that. */ - - /* Loop over all indexes on the relation */ - indexoidlist = RelationGetIndexList(rel); - - foreach(indexoidscan, indexoidlist) + if (!recurse) { - Oid indexoid = lfirst_oid(indexoidscan); - HeapTuple indexTuple; - Form_pg_index indexStruct; - int i; + if (rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE) + { + PartitionDesc partdesc; - indexTuple = SearchSysCache1(INDEXRELID, ObjectIdGetDatum(indexoid)); - if (!HeapTupleIsValid(indexTuple)) - elog(ERROR, "cache lookup failed for index %u", indexoid); - indexStruct = (Form_pg_index) GETSTRUCT(indexTuple); + partdesc = RelationGetPartitionDesc(rel, true); - /* - * If the index is not a primary key or an index used as replica - * identity, skip the check. - */ - if (indexStruct->indisprimary || indexStruct->indisreplident) + 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) { - /* - * 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 (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))); - } - } + 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.")); } - - ReleaseSysCache(indexTuple); } - list_free(indexoidlist); - - /* If rel is partition, shouldn't drop NOT NULL if parent has the same */ + /* + * 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); - Relation parent = table_open(parentId, AccessShareLock); - TupleDesc tupDesc = RelationGetDescr(parent); - AttrNumber parent_attnum; + Oid parentId = get_partition_parent(RelationGetRelid(rel), false); + Relation parent = table_open(parentId, AccessShareLock); + TupleDesc tupDesc = RelationGetDescr(parent); + AttrNumber parent_attnum; parent_attnum = get_attnum(parentId, colName); if (TupleDescAttr(tupDesc, parent_attnum - 1)->attnotnull) @@ -7428,22 +7605,33 @@ ATExecDropNotNull(Relation rel, const char *colName, LOCKMODE lockmode) } /* - * Okay, actually perform the catalog change ... if needed + * Find the constraint that makes this column NOT NULL. */ - if (attTup->attnotnull) + conTup = findNotNullConstraint(rel, colName); + if (conTup == NULL) { - attTup->attnotnull = false; + Bitmapset *pkcols; - CatalogTupleUpdate(attr_rel, &tuple->t_self, tuple); + /* + * There's no NOT NULL constraint, so throw an error. If the column + * is in a primary key, we can throw a specific error. Otherwise, + * this is unexpected. + */ + 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)); - ObjectAddressSubSet(address, RelationRelationId, - RelationGetRelid(rel), attnum); + /* this shouldn't happen */ + elog(ERROR, "no NOT NULL constraint found to drop"); } - else - address = InvalidObjectAddress; - InvokeObjectPostAlterHook(RelationRelationId, - RelationGetRelid(rel), attnum); + dropconstraint_internal(rel, conTup, DROP_RESTRICT, recurse, false, + false, NULL, lockmode); + + heap_freetuple(conTup); table_close(attr_rel, RowExclusiveLock); @@ -7451,102 +7639,126 @@ ATExecDropNotNull(Relation rel, const char *colName, LOCKMODE lockmode) } /* - * ALTER TABLE ALTER COLUMN SET NOT NULL + * 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. */ - static void -ATPrepSetNotNull(List **wqueue, Relation rel, - AlterTableCmd *cmd, bool recurse, bool recursing, - LOCKMODE lockmode, AlterTableUtilityContext *context) +set_attnotnull(List **wqueue, Relation rel, AttrNumber attnum, bool recurse, + LOCKMODE lockmode) { - /* - * If we're already recursing, there's nothing to do; the topmost - * invocation of ATSimpleRecursion already visited all children. - */ - if (recursing) - return; + HeapTuple tuple; + Form_pg_attribute attForm; + List *children; + ListCell *lc; - /* - * 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) + 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) { - HeapTuple tuple; - bool attnotnull; + Relation attr_rel; - tuple = SearchSysCacheAttName(RelationGetRelid(rel), cmd->name); + attr_rel = table_open(AttributeRelationId, RowExclusiveLock); - /* 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)))); + attForm->attnotnull = true; + CatalogTupleUpdate(attr_rel, &tuple->t_self, tuple); - attnotnull = ((Form_pg_attribute) GETSTRUCT(tuple))->attnotnull; - ReleaseSysCache(tuple); - if (attnotnull) - return; + table_close(attr_rel, RowExclusiveLock); + + /* + * And set up for existing values to be checked, unless another constraint + * already proves this. + */ + if (wqueue && !NotNullImpliedByRelConstraints(rel, attForm)) + { + AlteredTableInfo *tab; + + tab = ATGetQueueEntry(wqueue, rel); + tab->verify_new_notnull = true; + } } - /* - * 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) + /* if no recursion is desired, we're done */ + if (!recurse) + return; + + children = find_inheritance_children(RelationGetRelid(rel), lockmode); + foreach(lc, children) { - AlterTableCmd *newcmd = makeNode(AlterTableCmd); + Oid childrelid = lfirst_oid(lc); + Relation childrel; + AttrNumber childattno; + + /* find_inheritance_children already got lock */ + childrel = table_open(childrelid, NoLock); + CheckTableNotInUse(childrel, "ALTER TABLE"); - newcmd->subtype = AT_CheckNotNull; - newcmd->name = pstrdup(cmd->name); - ATSimpleRecursion(wqueue, rel, newcmd, true, lockmode, context); + childattno = get_attnum(RelationGetRelid(childrel), + get_attname(RelationGetRelid(rel), attnum, + false)); + set_attnotnull(wqueue, childrel, childattno, + recurse, lockmode); + table_close(childrel, NoLock); } - else - ATSimpleRecursion(wqueue, rel, cmd, recurse, lockmode, context); } /* - * Return the address of the modified column. If the column was already NOT - * NULL, InvalidObjectAddress is returned. + * 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. */ static ObjectAddress -ATExecSetNotNull(AlteredTableInfo *tab, Relation rel, - const char *colName, LOCKMODE lockmode) +ATExecSetNotNull(List **wqueue, Relation rel, char *conName, char *colName, + bool recurse, bool recursing, List **readyRels, + LOCKMODE lockmode) { HeapTuple tuple; + Relation constr_rel; + ScanKeyData skey; + SysScanDesc conscan; AttrNumber attnum; - Relation attr_rel; ObjectAddress address; + Constraint *constraint; + CookedConstraint *ccon; + List *cooked; + List *ready = NIL; /* - * lookup the attribute + * 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. */ - attr_rel = table_open(AttributeRelationId, RowExclusiveLock); + if (readyRels == NULL) + readyRels = &ready; + if (list_member_oid(*readyRels, RelationGetRelid(rel))) + return InvalidObjectAddress; + *readyRels = lappend_oid(*readyRels, RelationGetRelid(rel)); - tuple = SearchSysCacheCopyAttName(RelationGetRelid(rel), colName); + /* 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); + } - if (!HeapTupleIsValid(tuple)) + attnum = get_attnum(RelationGetRelid(rel), colName); + if (attnum == InvalidAttrNumber) 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, @@ -7554,43 +7766,168 @@ ATExecSetNotNull(AlteredTableInfo *tab, Relation rel, errmsg("cannot alter system column \"%s\"", colName))); - /* - * Okay, actually perform the catalog change ... if needed - */ - if (!((Form_pg_attribute) GETSTRUCT(tuple))->attnotnull) + /* 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))) { - ((Form_pg_attribute) GETSTRUCT(tuple))->attnotnull = true; + Form_pg_constraint conForm = (Form_pg_constraint) GETSTRUCT(tuple); + bool changed = false; + HeapTuple copytup; - CatalogTupleUpdate(attr_rel, &tuple->t_self, tuple); + if (conForm->contype != CONSTRAINT_NOTNULL) + continue; + + if (extractNotNullColumn(tuple) != attnum) + continue; + + copytup = heap_copytuple(tuple); + conForm = (Form_pg_constraint) GETSTRUCT(copytup); /* - * 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 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. */ - if (!tab->verify_new_notnull && - !NotNullImpliedByRelConstraints(rel, (Form_pg_attribute) GETSTRUCT(tuple))) + if (recursing) { - /* Tell Phase 3 it needs to test the constraint */ - tab->verify_new_notnull = true; + conForm->coninhcount++; + changed = true; + } + else if (!conForm->conislocal) + { + conForm->conislocal = true; + changed = true; } - ObjectAddressSubSet(address, RelationRelationId, - RelationGetRelid(rel), attnum); + if (changed) + { + CatalogTupleUpdate(constr_rel, ©tup->t_self, copytup); + ObjectAddressSet(address, ConstraintRelationId, conForm->oid); + } + + systable_endscan(conscan); + table_close(constr_rel, RowExclusiveLock); + + if (changed) + return address; + else + return InvalidObjectAddress; } - else - address = InvalidObjectAddress; - InvokeObjectPostAlterHook(RelationRelationId, - RelationGetRelid(rel), attnum); + systable_endscan(conscan); + table_close(constr_rel, RowExclusiveLock); - table_close(attr_rel, RowExclusiveLock); + /* + * If we're asked not to recurse, and children exist, raise an error. + */ + 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("cannot add constraint to only the partitioned table when partitions exist"), + errhint("Do not specify the ONLY keyword.")); + else + ereport(ERROR, + errcode(ERRCODE_INVALID_TABLE_DEFINITION), + errmsg("cannot add constraint only to table with inheritance children"), + errhint("Do not specify the ONLY keyword.")); + } + + /* + * 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->colname = colName; + 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); + + /* + * 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); + } + } return address; } /* + * ALTER TABLE ALTER COLUMN SET ATTNOTNULL + * + * This doesn't exist in the grammar; it's used when creating a + * primary key and the column is not already marked attnotnull. + */ +static void +ATExecSetAttNotNull(List **wqueue, Relation rel, + const char *colName, LOCKMODE lockmode) +{ + AttrNumber attnum; + + attnum = get_attnum(RelationGetRelid(rel), colName); + if (attnum == InvalidAttrNumber) /* XXX should not happen .. elog? */ + ereport(ERROR, + errcode(ERRCODE_UNDEFINED_COLUMN), + errmsg("column \"%s\" of relation \"%s\" does not exist", + colName, RelationGetRelationName(rel))); + + set_attnotnull(wqueue, rel, attnum, false, lockmode); +} + +/* * ALTER TABLE ALTER COLUMN CHECK NOT NULL * * This doesn't exist in the grammar, but we generate AT_CheckNotNull @@ -8872,13 +9209,14 @@ ATExecAddConstraint(List **wqueue, AlteredTableInfo *tab, Relation rel, Assert(IsA(newConstraint, Constraint)); /* - * 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. + * Currently, we only expect to see CONSTR_CHECK, CONSTR_NOTNULL and + * CONSTR_FOREIGN nodes arriving here (see the preprocessing done in + * parse_utilcmd.c). */ switch (newConstraint->contype) { case CONSTR_CHECK: + case CONSTR_NOTNULL: address = ATAddCheckConstraint(wqueue, tab, rel, newConstraint, recurse, false, is_readd, @@ -8963,9 +9301,9 @@ ChooseForeignKeyConstraintNameAddition(List *colnames) } /* - * 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. + * 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. * * Subroutine for ATExecAddConstraint. * @@ -9018,7 +9356,7 @@ ATAddCheckConstraint(List **wqueue, AlteredTableInfo *tab, Relation rel, { CookedConstraint *ccon = (CookedConstraint *) lfirst(lcon); - if (!ccon->skip_validation) + if (!ccon->skip_validation && ccon->contype != CONSTR_NOTNULL) { NewConstraint *newcon; @@ -9034,6 +9372,14 @@ ATAddCheckConstraint(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); } @@ -11958,16 +12304,11 @@ ATExecDropConstraint(Relation rel, const char *constrName, bool recurse, bool recursing, bool missing_ok, LOCKMODE lockmode) { - List *children; - ListCell *child; 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) @@ -11996,80 +12337,258 @@ ATExecDropConstraint(Relation rel, const char *constrName, /* There can be at most one matching row */ if (HeapTupleIsValid(tuple = systable_getnext(scan))) { - ObjectAddress conobj; + dropconstraint_internal(rel, tuple, behavior, recurse, recursing, + missing_ok, NULL, lockmode); + found = true; + } - con = (Form_pg_constraint) GETSTRUCT(tuple); + systable_endscan(scan); - /* Don't drop inherited constraints */ - if (con->coninhcount > 0 && !recursing) + if (!found) + { + if (!missing_ok) ereport(ERROR, - (errcode(ERRCODE_INVALID_TABLE_DEFINITION), - errmsg("cannot drop inherited constraint \"%s\" of relation \"%s\"", - constrName, RelationGetRelationName(rel)))); + 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))); + } - is_no_inherit_constraint = con->connoinherit; - contype = con->contype; + table_close(conrel, RowExclusiveLock); +} - /* - * 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; +/* + * 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; + ListCell *child; + bool is_no_inherit_constraint = false; + bool dropping_pk = false; + char *constrName; + List *unconstrained_cols = NIL; + char *colname; /* to match NOT NULL constraints when recursing */ + List *ready = NIL; + + if (readyRels == NULL) + readyRels = &ready; + if (list_member_oid(*readyRels, RelationGetRelid(rel))) + return InvalidObjectAddress; + *readyRels = lappend_oid(*readyRels, RelationGetRelid(rel)); - /* Must match lock taken by RemoveTriggerById: */ - frel = table_open(con->confrelid, AccessExclusiveLock); - CheckTableNotInUse(frel, "ALTER TABLE"); - table_close(frel, NoLock); - } + conrel = table_open(ConstraintRelationId, RowExclusiveLock); - /* - * Perform the actual constraint deletion - */ - conobj.classId = ConstraintRelationId; - conobj.objectId = con->oid; - conobj.objectSubId = 0; + con = (Form_pg_constraint) GETSTRUCT(constraintTup); + constrName = NameStr(con->conname); - performDeletion(&conobj, behavior, 0); + /* + * If the constraint is marked conislocal and is also inherited, then we + * just set conislocal false and we're done. The constraint doesn't go + * away, and we don't modify any children. + */ + if (con->conislocal && con->coninhcount > 0) + { + HeapTuple copytup; - found = true; + /* make a copy we can scribble on */ + copytup = heap_copytuple(constraintTup); + con = (Form_pg_constraint) GETSTRUCT(copytup); + con->conislocal = false; + CatalogTupleUpdate(conrel, ©tup->t_self, copytup); + + table_close(conrel, RowExclusiveLock); + + ObjectAddressSet(conobj, ConstraintRelationId, con->oid); + return conobj; } - systable_endscan(scan); + /* Don't drop inherited constraints */ + if (con->coninhcount > 0 && !recursing) + ereport(ERROR, + (errcode(ERRCODE_INVALID_TABLE_DEFINITION), + errmsg("cannot drop inherited constraint \"%s\" of relation \"%s\"", + constrName, RelationGetRelationName(rel)))); - if (!found) + /* + * 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) { - if (!missing_ok) - { - ereport(ERROR, - (errcode(ERRCODE_UNDEFINED_OBJECT), - errmsg("constraint \"%s\" of relation \"%s\" does not exist", - constrName, RelationGetRelationName(rel)))); - } - else + AttrNumber colnum = extractNotNullColumn(constraintTup); + + if (colnum != InvalidAttrNumber) + unconstrained_cols = list_make1_int(colnum); + } + else if (con->contype == CONSTRAINT_PRIMARY) + { + Datum adatum; + ArrayType *arr; + int numkeys; + bool isNull; + int16 *attnums; + + dropping_pk = true; + + 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); + + for (int i = 0; i < numkeys; i++) + unconstrained_cols = lappend_int(unconstrained_cols, attnums[i]); + } + + is_no_inherit_constraint = con->connoinherit; + + /* + * 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; + + /* Must match lock taken by RemoveTriggerById: */ + frel = table_open(con->confrelid, AccessExclusiveLock); + CheckTableNotInUse(frel, "ALTER TABLE"); + table_close(frel, NoLock); + } + + /* + * Perform the actual constraint deletion + */ + ObjectAddressSet(conobj, ConstraintRelationId, con->oid); + performDeletion(&conobj, behavior, 0); + + /* + * 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) + { + Relation attrel; + Bitmapset *pkcols; + Bitmapset *ircols; + ListCell *lc; + + /* Make the above deletion 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(lc, unconstrained_cols) { - ereport(NOTICE, - (errmsg("constraint \"%s\" of relation \"%s\" does not exist, skipping", - constrName, RelationGetRelationName(rel)))); - table_close(conrel, RowExclusiveLock); - return; + AttrNumber attnum = lfirst_int(lc); + HeapTuple atttup; + HeapTuple contup; + Form_pg_attribute attForm; + + /* + * Obtain pg_attribute tuple and verify conditions on it. We use + * a copy we can scribble on. + */ + atttup = SearchSysCacheCopyAttNum(RelationGetRelid(rel), attnum); + if (!HeapTupleIsValid(atttup)) + elog(ERROR, "cache lookup failed for column %d", attnum); + attForm = (Form_pg_attribute) GETSTRUCT(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, and reset attnotnull if none. + * + * However, if this is a generated identity column, abort the + * whole thing with a specific error message, because the + * constraint is required in that case. + */ + contup = findNotNullConstraintAttnum(rel, attnum); + if (contup || + bms_is_member(attnum - FirstLowInvalidHeapAttributeNumber, + pkcols)) + continue; + + /* + * It's not valid to drop the last NOT NULL constraint for a + * GENERATED AS IDENTITY column. + */ + if (attForm->attidentity) + 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 last NOT NULL constraint for the + * replica identity either. XXX make exception for FULL? + */ + if (bms_is_member(lfirst_int(lc) - FirstLowInvalidHeapAttributeNumber, ircols)) + ereport(ERROR, + errcode(ERRCODE_INVALID_TABLE_DEFINITION), + errmsg("column \"%s\" is in index used as replica identity", + get_attname(RelationGetRelid(rel), lfirst_int(lc), false))); + + /* Reset attnotnull */ + if (attForm->attnotnull) + { + attForm->attnotnull = false; + CatalogTupleUpdate(attrel, &atttup->t_self, atttup); + } } + table_close(attrel, RowExclusiveLock); } /* * For partitioned tables, non-CHECK inherited constraints are dropped via * the dependency mechanism, so we're done here. */ - if (contype != CONSTRAINT_CHECK && + if (con->contype != CONSTRAINT_CHECK && rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE) { table_close(conrel, RowExclusiveLock); - return; + return conobj; } /* @@ -12094,50 +12613,104 @@ ATExecDropConstraint(Relation rel, const char *constrName, errmsg("cannot remove constraint from only the partitioned table when partitions exist"), errhint("Do not specify the ONLY keyword."))); + /* For NOT NULL constraints we recurse by column name */ + if (con->contype == CONSTRAINT_NOTNULL) + colname = NameStr(TupleDescAttr(RelationGetDescr(rel), + linitial_int(unconstrained_cols) - 1)->attname); + else + colname = NULL; /* keep compiler quiet */ + foreach(child, children) { Oid childrelid = lfirst_oid(child); Relation childrel; + HeapTuple tuple; + Form_pg_constraint childcon; HeapTuple copy_tuple; + SysScanDesc scan; + ScanKeyData skey[3]; + + if (list_member_oid(*readyRels, childrelid)) + continue; /* child already processed */ /* find_inheritance_children already got lock */ childrel = table_open(childrelid, NoLock); CheckTableNotInUse(childrel, "ALTER TABLE"); - 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)))); + /* + * We search for NOT NULL constraint by column number, and other + * constraints by name. + */ + if (con->contype == CONSTRAINT_NOTNULL) + { + bool found = false; + AttrNumber child_colnum; + HeapTuple child_tup; + + child_colnum = get_attnum(RelationGetRelid(childrel), colname); + ScanKeyInit(&skey[0], + Anum_pg_constraint_conrelid, + BTEqualStrategyNumber, F_OIDEQ, + ObjectIdGetDatum(childrelid)); + scan = systable_beginscan(conrel, ConstraintRelidTypidNameIndexId, + true, NULL, 1, skey); + while (HeapTupleIsValid(child_tup = systable_getnext(scan))) + { + Form_pg_constraint constr = (Form_pg_constraint) GETSTRUCT(child_tup); + AttrNumber constr_colnum; - copy_tuple = heap_copytuple(tuple); + if (constr->contype != CONSTRAINT_NOTNULL) + continue; + constr_colnum = extractNotNullColumn(child_tup); + if (constr_colnum != child_colnum) + continue; - systable_endscan(scan); + found = true; + break; /* found it */ + } + if (!found) /* shouldn't happen? */ + elog(ERROR, "failed to find NOT NULL constraint for column \"%s\" in table \"%s\"", + colname, RelationGetRelationName(childrel)); + + copy_tuple = heap_copytuple(child_tup); + systable_endscan(scan); + } + else + { + 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)))); + copy_tuple = heap_copytuple(tuple); + systable_endscan(scan); + } - con = (Form_pg_constraint) GETSTRUCT(copy_tuple); + childcon = (Form_pg_constraint) GETSTRUCT(copy_tuple); - /* Right now only CHECK constraints can be inherited */ - if (con->contype != CONSTRAINT_CHECK) - elog(ERROR, "inherited constraint is not a CHECK constraint"); + /* 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"); - if (con->coninhcount <= 0) /* shouldn't happen */ + if (childcon->coninhcount <= 0) /* shouldn't happen */ elog(ERROR, "relation %u has non-inherited constraint \"%s\"", childrelid, constrName); @@ -12147,17 +12720,17 @@ ATExecDropConstraint(Relation rel, const char *constrName, * If the child constraint has other definition sources, just * decrement its inheritance count; if not, recurse to delete it. */ - if (con->coninhcount == 1 && !con->conislocal) + if (childcon->coninhcount == 1 && !childcon->conislocal) { /* Time to delete this child constraint, too */ - ATExecDropConstraint(childrel, constrName, behavior, - true, true, - false, lockmode); + dropconstraint_internal(childrel, copy_tuple, behavior, + recurse, true, missing_ok, readyRels, + lockmode); } else { /* Child constraint must survive my deletion */ - con->coninhcount--; + childcon->coninhcount--; CatalogTupleUpdate(conrel, ©_tuple->t_self, copy_tuple); /* Make update visible */ @@ -12171,8 +12744,8 @@ ATExecDropConstraint(Relation rel, const char *constrName, * need to mark the inheritors' constraints as locally defined * rather than inherited. */ - con->coninhcount--; - con->conislocal = true; + childcon->coninhcount--; + childcon->conislocal = true; CatalogTupleUpdate(conrel, ©_tuple->t_self, copy_tuple); @@ -12186,6 +12759,8 @@ ATExecDropConstraint(Relation rel, const char *constrName, } table_close(conrel, RowExclusiveLock); + + return conobj; } /* @@ -13511,10 +14086,10 @@ ATPostAlterTypeParse(Oid oldId, Oid oldRelId, Oid refRelId, char *cmd, NIL, con->conname); } - else if (cmd->subtype == AT_SetNotNull) + else if (cmd->subtype == AT_SetAttNotNull) { /* - * The parser will create AT_SetNotNull subcommands for + * The parser will create AT_AttSetNotNull 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 @@ -15258,6 +15833,7 @@ MergeConstraintsIntoExisting(Relation child_rel, Relation parent_rel) SysScanDesc parent_scan; ScanKeyData parent_key; HeapTuple parent_tuple; + Oid parent_relid = RelationGetRelid(parent_rel); bool child_is_partition = false; catalog_relation = table_open(ConstraintRelationId, RowExclusiveLock); @@ -15271,7 +15847,7 @@ MergeConstraintsIntoExisting(Relation child_rel, Relation parent_rel) ScanKeyInit(&parent_key, Anum_pg_constraint_conrelid, BTEqualStrategyNumber, F_OIDEQ, - ObjectIdGetDatum(RelationGetRelid(parent_rel))); + ObjectIdGetDatum(parent_relid)); parent_scan = systable_beginscan(catalog_relation, ConstraintRelidTypidNameIndexId, true, NULL, 1, &parent_key); @@ -15283,7 +15859,8 @@ MergeConstraintsIntoExisting(Relation child_rel, Relation parent_rel) HeapTuple child_tuple; bool found = false; - if (parent_con->contype != CONSTRAINT_CHECK) + if (parent_con->contype != CONSTRAINT_CHECK && + parent_con->contype != CONSTRAINT_NOTNULL) continue; /* if the parent's constraint is marked NO INHERIT, it's not inherited */ @@ -15303,14 +15880,30 @@ 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 != CONSTRAINT_CHECK) + if (child_con->contype != parent_con->contype) continue; - if (strcmp(NameStr(parent_con->conname), + /* + * CHECK constraint are matched by name, NOT NULL ones by + * attribute number + */ + if (child_con->contype == CONSTRAINT_CHECK && + 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 (!constraints_equivalent(parent_tuple, child_tuple, tuple_desc)) + if (child_con->contype == CONSTRAINT_CHECK && + !constraints_equivalent(parent_tuple, child_tuple, tuple_desc)) ereport(ERROR, (errcode(ERRCODE_DATATYPE_MISMATCH), errmsg("child table \"%s\" has different definition for check constraint \"%s\"", @@ -15518,6 +16111,7 @@ RemoveInheritance(Relation child_rel, Relation parent_rel, bool expect_detached) HeapTuple attributeTuple, constraintTuple; List *connames; + List *nncolumns; bool found; bool child_is_partition = false; @@ -15588,6 +16182,8 @@ 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], @@ -15598,6 +16194,7 @@ RemoveInheritance(Relation child_rel, Relation parent_rel, bool expect_detached) true, NULL, 1, key); connames = NIL; + nncolumns = NIL; while (HeapTupleIsValid(constraintTuple = systable_getnext(scan))) { @@ -15605,6 +16202,8 @@ 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); @@ -15620,21 +16219,40 @@ 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; + bool match = false; ListCell *lc; - if (con->contype != CONSTRAINT_CHECK) - continue; - - match = false; - foreach(lc, connames) + /* + * Match CHECK constraints by name, NOT NULL constraints by column + * number, and ignore all others. + */ + if (con->contype == CONSTRAINT_CHECK) { - if (strcmp(NameStr(con->conname), (char *) lfirst(lc)) == 0) + foreach(lc, connames) { - match = true; - break; + 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); + + foreach(lc, nncolumns) + { + if (lfirst_int(lc) == child_attno) + { + match = true; + break; + } + } + } + else + continue; if (match) { @@ -17925,7 +18543,7 @@ ATExecAttachPartition(List **wqueue, Relation rel, PartitionCmd *cmd, StorePartitionBound(attachrel, rel, cmd->bound); /* Ensure there exists a correct set of indexes in the partition. */ - AttachPartitionEnsureIndexes(rel, attachrel); + AttachPartitionEnsureIndexes(wqueue, rel, attachrel); /* and triggers */ CloneRowTriggersToPartition(rel, attachrel); @@ -18038,13 +18656,12 @@ ATExecAttachPartition(List **wqueue, Relation rel, PartitionCmd *cmd, * partitioned table. */ static void -AttachPartitionEnsureIndexes(Relation rel, Relation attachrel) +AttachPartitionEnsureIndexes(List **wqueue, Relation rel, Relation attachrel) { List *idxes; List *attachRelIdxs; Relation *attachrelIdxRels; IndexInfo **attachInfos; - int i; ListCell *cell; MemoryContext cxt; MemoryContext oldcxt; @@ -18060,14 +18677,13 @@ AttachPartitionEnsureIndexes(Relation rel, Relation attachrel) attachInfos = palloc(sizeof(IndexInfo *) * list_length(attachRelIdxs)); /* Build arrays of all existing indexes and their IndexInfos */ - i = 0; foreach(cell, attachRelIdxs) { Oid cldIdxId = lfirst_oid(cell); + int i = foreach_current_index(cell); attachrelIdxRels[i] = index_open(cldIdxId, AccessShareLock); attachInfos[i] = BuildIndexInfo(attachrelIdxRels[i]); - i++; } /* @@ -18133,7 +18749,7 @@ AttachPartitionEnsureIndexes(Relation rel, Relation attachrel) * the first matching, unattached one we find, if any, as partition of * the parent index. If we find one, we're done. */ - for (i = 0; i < list_length(attachRelIdxs); i++) + for (int i = 0; i < list_length(attachRelIdxs); i++) { Oid cldIdxId = RelationGetRelid(attachrelIdxRels[i]); Oid cldConstrOid = InvalidOid; @@ -18189,6 +18805,28 @@ AttachPartitionEnsureIndexes(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, @@ -18201,7 +18839,7 @@ AttachPartitionEnsureIndexes(Relation rel, Relation attachrel) out: /* Clean up. */ - for (i = 0; i < list_length(attachRelIdxs); i++) + for (int i = 0; i < list_length(attachRelIdxs); i++) index_close(attachrelIdxRels[i], AccessShareLock); MemoryContextSwitchTo(oldcxt); MemoryContextDelete(cxt); @@ -19102,6 +19740,13 @@ 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)) @@ -19239,6 +19884,29 @@ 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/nodes/outfuncs.c b/src/backend/nodes/outfuncs.c index ba00b99249f..d6d67c90839 100644 --- a/src/backend/nodes/outfuncs.c +++ b/src/backend/nodes/outfuncs.c @@ -717,6 +717,10 @@ _outConstraint(StringInfo str, const Constraint *node) case CONSTR_NOTNULL: appendStringInfoString(str, "NOT_NULL"); + WRITE_BOOL_FIELD(is_no_inherit); + WRITE_STRING_FIELD(colname); + WRITE_BOOL_FIELD(skip_validation); + WRITE_BOOL_FIELD(initially_valid); break; case CONSTR_DEFAULT: diff --git a/src/backend/nodes/readfuncs.c b/src/backend/nodes/readfuncs.c index 597e5b3ea8b..4c200485f3a 100644 --- a/src/backend/nodes/readfuncs.c +++ b/src/backend/nodes/readfuncs.c @@ -390,10 +390,16 @@ _readConstraint(void) switch (local_node->contype) { case CONSTR_NULL: - case CONSTR_NOTNULL: /* no extra fields */ break; + case CONSTR_NOTNULL: + READ_BOOL_FIELD(is_no_inherit); + READ_STRING_FIELD(colname); + READ_BOOL_FIELD(skip_validation); + READ_BOOL_FIELD(initially_valid); + break; + case CONSTR_DEFAULT: READ_NODE_FIELD(raw_expr); READ_STRING_FIELD(cooked_expr); diff --git a/src/backend/optimizer/util/plancat.c b/src/backend/optimizer/util/plancat.c index e3824efe9b5..75d84459141 100644 --- a/src/backend/optimizer/util/plancat.c +++ b/src/backend/optimizer/util/plancat.c @@ -1644,6 +1644,8 @@ 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 acf6cf48669..86692bf3959 100644 --- a/src/backend/parser/gram.y +++ b/src/backend/parser/gram.y @@ -4099,6 +4099,19 @@ 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->colname = $3; + processCASbits($4, @4, "NOT NULL", + NULL, NULL, NULL, + &n->is_no_inherit, yyscanner); + n->initially_valid = !n->skip_validation; + $$ = (Node *) n; + } | UNIQUE opt_unique_null_treatment '(' columnList ')' opt_c_include opt_definition OptConsTableSpace ConstraintAttributeSpec { diff --git a/src/backend/parser/parse_utilcmd.c b/src/backend/parser/parse_utilcmd.c index b0f6fe4fa61..15d8f3a1a0b 100644 --- a/src/backend/parser/parse_utilcmd.c +++ b/src/backend/parser/parse_utilcmd.c @@ -83,6 +83,7 @@ typedef struct List *ckconstraints; /* CHECK constraints */ List *fkconstraints; /* FOREIGN KEY constraints */ List *ixconstraints; /* index-creating constraints */ + List *nnconstraints; /* NOT NULL constraints */ List *likeclauses; /* LIKE clauses that need post-processing */ List *extstats; /* cloned extended statistics */ List *blist; /* "before list" of things to do before @@ -244,6 +245,7 @@ transformCreateStmt(CreateStmt *stmt, const char *queryString) cxt.ckconstraints = NIL; cxt.fkconstraints = NIL; cxt.ixconstraints = NIL; + cxt.nnconstraints = NIL; cxt.likeclauses = NIL; cxt.extstats = NIL; cxt.blist = NIL; @@ -348,6 +350,7 @@ 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); @@ -537,6 +540,7 @@ 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); @@ -634,10 +638,8 @@ transformColumnDefinition(CreateStmtContext *cxt, ColumnDef *column) constraint->cooked_expr = NULL; column->constraints = lappend(column->constraints, constraint); - constraint = makeNode(Constraint); - constraint->contype = CONSTR_NOTNULL; - constraint->location = -1; - column->constraints = lappend(column->constraints, constraint); + /* have a NOT NULL constraint added later */ + need_notnull = true; } /* Process column constraints, if any... */ @@ -655,7 +657,7 @@ transformColumnDefinition(CreateStmtContext *cxt, ColumnDef *column) switch (constraint->contype) { case CONSTR_NULL: - if (saw_nullable && column->is_not_null) + if ((saw_nullable && column->is_not_null) || need_notnull) ereport(ERROR, (errcode(ERRCODE_SYNTAX_ERROR), errmsg("conflicting NULL/NOT NULL declarations for column \"%s\" of table \"%s\"", @@ -667,15 +669,58 @@ transformColumnDefinition(CreateStmtContext *cxt, ColumnDef *column) break; case CONSTR_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\"", - column->colname, cxt->relation->relname), - parser_errposition(cxt->pstate, - constraint->location))); - column->is_not_null = true; - saw_nullable = true; + + /* + * For NOT NULL declarations, we need to mark the column as + * not nullable, and set things up to have a CHECK constraint + * created. Also, duplicate NOT NULL declarations are not + * allowed. + */ + if (saw_nullable) + { + if (!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))); + else + ereport(ERROR, + errcode(ERRCODE_SYNTAX_ERROR), + errmsg("redundant NOT NULL declarations for column \"%s\" of table \"%s\"", + column->colname, cxt->relation->relname), + parser_errposition(cxt->pstate, + constraint->location)); + } + + /* + * If this is the first time we see this column being marked + * not null, keep track to later add a NOT NULL constraint. + */ + if (!column->is_not_null) + { + Constraint *notnull; + + column->is_not_null = true; + saw_nullable = true; + + notnull = makeNode(Constraint); + notnull->contype = CONSTR_NOTNULL; + notnull->conname = constraint->conname; + notnull->deferrable = false; + notnull->initdeferred = false; + notnull->location = -1; + notnull->colname = column->colname; + notnull->skip_validation = false; + notnull->initially_valid = true; + + cxt->nnconstraints = lappend(cxt->nnconstraints, notnull); + + /* Don't need this anymore, if we had it */ + need_notnull = false; + } + break; case CONSTR_DEFAULT: @@ -725,16 +770,19 @@ transformColumnDefinition(CreateStmtContext *cxt, ColumnDef *column) column->identity = constraint->generated_when; saw_identity = true; - /* An identity column is implicitly NOT NULL */ - if (saw_nullable && !column->is_not_null) + /* + * 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) 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; } @@ -758,6 +806,11 @@ transformColumnDefinition(CreateStmtContext *cxt, ColumnDef *column) case CONSTR_CHECK: cxt->ckconstraints = lappend(cxt->ckconstraints, constraint); + + /* + * XXX If the user says CHECK (IS NOT NULL), should we turn + * that into a regular NOT NULL constraint? + */ break; case CONSTR_PRIMARY: @@ -841,6 +894,29 @@ 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->colname = 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. */ @@ -915,6 +991,10 @@ transformTableConstraint(CreateStmtContext *cxt, Constraint *constraint) cxt->ckconstraints = lappend(cxt->ckconstraints, constraint); break; + case CONSTR_NOTNULL: + cxt->nnconstraints = lappend(cxt->nnconstraints, constraint); + break; + case CONSTR_FOREIGN: if (cxt->isforeign) ereport(ERROR, @@ -926,7 +1006,6 @@ transformTableConstraint(CreateStmtContext *cxt, Constraint *constraint) break; case CONSTR_NULL: - case CONSTR_NOTNULL: case CONSTR_DEFAULT: case CONSTR_ATTR_DEFERRABLE: case CONSTR_ATTR_NOT_DEFERRABLE: @@ -962,6 +1041,7 @@ transformTableLikeClause(CreateStmtContext *cxt, TableLikeClause *table_like_cla AclResult aclresult; char *comment; ParseCallbackState pcbstate; + bool process_notnull_constraints; setup_parser_errposition_callback(&pcbstate, cxt->pstate, table_like_clause->relation->location); @@ -1043,6 +1123,8 @@ transformTableLikeClause(CreateStmtContext *cxt, TableLikeClause *table_like_cla def->inhcount = 0; def->is_local = true; def->is_not_null = attribute->attnotnull; + if (attribute->attnotnull) + process_notnull_constraints = true; def->is_from_type = false; def->storage = 0; def->raw_default = NULL; @@ -1124,14 +1206,19 @@ 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. Also, remember the relation OID so that + * 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 * expandTableLikeClause is certain to open the same table. */ - if (table_like_clause->options & + if ((table_like_clause->options & (CREATE_TABLE_LIKE_DEFAULTS | CREATE_TABLE_LIKE_GENERATED | CREATE_TABLE_LIKE_CONSTRAINTS | - CREATE_TABLE_LIKE_INDEXES)) + CREATE_TABLE_LIKE_INDEXES)) || + process_notnull_constraints) { table_like_clause->relationOid = RelationGetRelid(relation); cxt->likeclauses = lappend(cxt->likeclauses, table_like_clause); @@ -1203,6 +1290,7 @@ expandTableLikeClause(RangeVar *heapRel, TableLikeClause *table_like_clause) TupleConstr *constr; AttrMap *attmap; char *comment; + ListCell *lc; /* * Open the relation referenced by the LIKE clause. We should still have @@ -1383,6 +1471,20 @@ 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(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. @@ -2059,10 +2161,12 @@ transformIndexConstraints(CreateStmtContext *cxt) ListCell *lc; /* - * 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. + * 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 CHECK (IS 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. */ foreach(lc, cxt->ixconstraints) { @@ -2136,9 +2240,7 @@ transformIndexConstraints(CreateStmtContext *cxt) } /* - * 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. + * Now append all the IndexStmts to cxt->alist. */ cxt->alist = list_concat(cxt->alist, finalindexlist); } @@ -2146,12 +2248,10 @@ transformIndexConstraints(CreateStmtContext *cxt) /* * transformIndexConstraint * Transform one UNIQUE, PRIMARY KEY, or EXCLUDE constraint for - * transformIndexConstraints. + * transformIndexConstraints. An IndexStmt is returned. * - * 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. + * For a PRIMARY KEY constraint, we additionally force the columns to be + * marked as NOT NULL, without producing a CHECK (IS NOT NULL) constraint. */ static IndexStmt * transformIndexConstraint(Constraint *constraint, CreateStmtContext *cxt) @@ -2417,7 +2517,6 @@ 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; @@ -2438,13 +2537,14 @@ 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). + * 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. */ if (constraint->contype == CONSTR_PRIMARY && !column->is_from_type) { column->is_not_null = true; - forced_not_null = true; } } else if (SystemAttributeByName(key) != NULL) @@ -2487,14 +2587,6 @@ 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; } } @@ -2548,15 +2640,11 @@ transformIndexConstraint(Constraint *constraint, CreateStmtContext *cxt) iparam->nulls_ordering = SORTBY_NULLS_DEFAULT; index->indexParams = lappend(index->indexParams, iparam); - /* - * 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) + if (constraint->contype == CONSTR_PRIMARY) { AlterTableCmd *notnullcmd = makeNode(AlterTableCmd); - notnullcmd->subtype = AT_SetNotNull; + notnullcmd->subtype = AT_SetAttNotNull; notnullcmd->name = pstrdup(key); notnullcmds = lappend(notnullcmds, notnullcmd); } @@ -3328,6 +3416,7 @@ 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; @@ -3571,8 +3660,8 @@ transformAlterTableStmt(Oid relid, AlterTableStmt *stmt, /* * We assume here that cxt.alist contains only IndexStmts and possibly - * ALTER TABLE SET NOT NULL statements generated from primary key - * constraints. We absorb the subcommands of the latter directly. + * AT_SetAttNotNull statements generated from primary key constraints. + * We absorb the subcommands of the latter directly. */ if (IsA(istmt, IndexStmt)) { @@ -3600,14 +3689,21 @@ transformAlterTableStmt(Oid relid, AlterTableStmt *stmt, { newcmd = makeNode(AlterTableCmd); newcmd->subtype = AT_AddConstraint; - newcmd->def = (Node *) lfirst(l); + newcmd->def = (Node *) lfirst_node(Constraint, l); newcmds = lappend(newcmds, newcmd); } foreach(l, cxt.fkconstraints) { newcmd = makeNode(AlterTableCmd); newcmd->subtype = AT_AddConstraint; - newcmd->def = (Node *) lfirst(l); + 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); } diff --git a/src/backend/utils/adt/ruleutils.c b/src/backend/utils/adt/ruleutils.c index 461735e84f0..0242cf24b4e 100644 --- a/src/backend/utils/adt/ruleutils.c +++ b/src/backend/utils/adt/ruleutils.c @@ -2472,6 +2472,20 @@ pg_get_constraintdef_worker(Oid constraintId, bool fullCommand, conForm->connoinherit ? " NO INHERIT" : ""); break; } + case CONSTRAINT_NOTNULL: + { + 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"); + break; + } + case CONSTRAINT_TRIGGER: /* |