summaryrefslogtreecommitdiff
path: root/src/backend
diff options
context:
space:
mode:
Diffstat (limited to 'src/backend')
-rw-r--r--src/backend/catalog/heap.c529
-rw-r--r--src/backend/catalog/pg_constraint.c286
-rw-r--r--src/backend/commands/tablecmds.c1591
-rw-r--r--src/backend/nodes/outfuncs.c5
-rw-r--r--src/backend/nodes/readfuncs.c9
-rw-r--r--src/backend/optimizer/util/plancat.c2
-rw-r--r--src/backend/parser/gram.y19
-rw-r--r--src/backend/parser/parse_utilcmd.c277
-rw-r--r--src/backend/utils/adt/ruleutils.c14
-rw-r--r--src/backend/utils/cache/relcache.c34
10 files changed, 2179 insertions, 587 deletions
diff --git a/src/backend/catalog/heap.c b/src/backend/catalog/heap.c
index 70bc4c00068..b534da7d806 100644
--- a/src/backend/catalog/heap.c
+++ b/src/backend/catalog/heap.c
@@ -1679,7 +1679,7 @@ RemoveAttributeById(Oid relid, AttrNumber attnum)
*/
attStruct->atttypid = InvalidOid;
- /* Remove any NOT NULL constraint the column may have */
+ /* Remove any not-null constraint the column may have */
attStruct->attnotnull = false;
/* We don't want to keep stats for it anymore */
@@ -2148,6 +2148,53 @@ StoreRelCheck(Relation rel, const char *ccname, Node *expr,
}
/*
+ * Store a not-null constraint for the given relation
+ *
+ * The OID of the new constraint is returned.
+ */
+static Oid
+StoreRelNotNull(Relation rel, const char *nnname, AttrNumber attnum,
+ bool is_validated, bool is_local, int inhcount,
+ bool is_no_inherit)
+{
+ Oid constrOid;
+
+ constrOid =
+ CreateConstraintEntry(nnname,
+ RelationGetNamespace(rel),
+ CONSTRAINT_NOTNULL,
+ false,
+ false,
+ is_validated,
+ InvalidOid,
+ RelationGetRelid(rel),
+ &attnum,
+ 1,
+ 1,
+ InvalidOid, /* not a domain constraint */
+ InvalidOid, /* no associated index */
+ InvalidOid, /* Foreign key fields */
+ NULL,
+ NULL,
+ NULL,
+ NULL,
+ 0,
+ ' ',
+ ' ',
+ NULL,
+ 0,
+ ' ',
+ NULL, /* not an exclusion constraint */
+ NULL,
+ NULL,
+ is_local,
+ inhcount,
+ is_no_inherit,
+ false);
+ return constrOid;
+}
+
+/*
* Store defaults and constraints (passed as a list of CookedConstraint).
*
* Each CookedConstraint struct is modified to store the new catalog tuple OID.
@@ -2191,6 +2238,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);
@@ -2248,6 +2303,7 @@ AddRelationNewConstraints(Relation rel,
ParseNamespaceItem *nsitem;
int numchecks;
List *checknames;
+ List *nnnames;
ListCell *cell;
Node *expr;
CookedConstraint *cooked;
@@ -2333,130 +2389,196 @@ 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.
- */
- 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;
+ CookedConstraint *nncooked;
+ AttrNumber colnum;
+ char *nnname;
- vars = pull_var_clause(expr, 0);
+ /* Determine which column to modify */
+ colnum = get_attnum(RelationGetRelid(rel), strVal(linitial(cdef->keys)));
+ if (colnum == InvalidAttrNumber) /* shouldn't happen */
+ elog(ERROR, "cache lookup failed for attribute \"%s\" of relation %u",
+ strVal(linitial(cdef->keys)), RelationGetRelid(rel));
- /* eliminate duplicates */
- vars = list_union(NIL, vars);
+ /*
+ * If the column already has a not-null constraint, we need only
+ * update its catalog status and we're done.
+ */
+ if (AdjustNotNullInheritance1(RelationGetRelid(rel), colnum,
+ cdef->inhcount))
+ continue;
- if (list_length(vars) == 1)
- colname = get_attname(RelationGetRelid(rel),
- ((Var *) linitial(vars))->varattno,
- true);
+ /*
+ * If a constraint name is specified, check that it isn't already
+ * used. Otherwise, choose a non-conflicting one ourselves.
+ */
+ if (cdef->conname)
+ {
+ if (ConstraintNameIsUsed(CONSTRAINT_RELATION,
+ RelationGetRelid(rel),
+ cdef->conname))
+ ereport(ERROR,
+ errcode(ERRCODE_DUPLICATE_OBJECT),
+ errmsg("constraint \"%s\" for relation \"%s\" already exists",
+ cdef->conname, RelationGetRelationName(rel)));
+ nnname = cdef->conname;
+ }
else
- colname = NULL;
-
- ccname = ChooseConstraintName(RelationGetRelationName(rel),
- colname,
- "check",
- RelationGetNamespace(rel),
- checknames);
-
- /* save name for future checks */
- checknames = lappend(checknames, ccname);
+ nnname = ChooseConstraintName(RelationGetRelationName(rel),
+ strVal(linitial(cdef->keys)),
+ "not_null",
+ RelationGetNamespace(rel),
+ nnnames);
+ nnnames = lappend(nnnames, nnname);
+
+ constrOid =
+ StoreRelNotNull(rel, nnname, colnum,
+ cdef->initially_valid,
+ cdef->inhcount == 0,
+ cdef->inhcount,
+ cdef->is_no_inherit);
+
+ nncooked = (CookedConstraint *) palloc(sizeof(CookedConstraint));
+ nncooked->contype = CONSTR_NOTNULL;
+ nncooked->conoid = constrOid;
+ nncooked->name = nnname;
+ nncooked->attnum = colnum;
+ nncooked->expr = NULL;
+ nncooked->skip_validation = cdef->skip_validation;
+ nncooked->is_local = is_local;
+ nncooked->inhcount = cdef->inhcount;
+ nncooked->is_no_inherit = cdef->is_no_inherit;
+
+ cookedConstraints = lappend(cookedConstraints, nncooked);
}
-
- /*
- * 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);
}
/*
@@ -2626,6 +2748,211 @@ 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 when creating a new relation
+ *
+ * These come from two sources: the 'constraints' list (of Constraint) is
+ * specified directly by the user; the 'old_notnulls' list (of
+ * CookedConstraint) comes from inheritance. We create one constraint
+ * for each column, giving priority to user-specified ones, and setting
+ * inhcount according to how many parents cause each column to get a
+ * not-null constraint. If a user-specified name clashes with another
+ * user-specified name, an error is raised.
+ *
+ * Note that inherited constraints have two shapes: those coming from another
+ * not-null constraint in the parent, which have a name already, and those
+ * coming from a primary key in the parent, which don't. Any name specified
+ * in a parent is disregarded in case of a conflict.
+ *
+ * Returns a list of AttrNumber for columns that need to have the attnotnull
+ * flag set.
+ */
+List *
+AddRelationNotNullConstraints(Relation rel, List *constraints,
+ List *old_notnulls)
+{
+ List *givennames;
+ List *nnnames;
+ List *nncols = NIL;
+ ListCell *lc;
+
+ /*
+ * We track two lists of names: nnnames keeps all the constraint names,
+ * givennames tracks user-generated names. The distinction is important,
+ * because we must raise error for user-generated name conflicts, but for
+ * system-generated name conflicts we just generate another.
+ */
+ nnnames = NIL;
+ givennames = NIL;
+
+ /*
+ * First, create all not-null constraints that are directly specified by
+ * the user. Note that inheritance might have given us another source for
+ * each, so we must scan the old_notnulls list and increment inhcount for
+ * each element with identical attnum. We delete from there any element
+ * that we process.
+ */
+ foreach(lc, constraints)
+ {
+ Constraint *constr = lfirst_node(Constraint, lc);
+ AttrNumber attnum;
+ char *conname;
+ bool is_local = true;
+ int inhcount = 0;
+ ListCell *lc2;
+
+ Assert(constr->contype == CONSTR_NOTNULL);
+
+ attnum = get_attnum(RelationGetRelid(rel),
+ strVal(linitial(constr->keys)));
+
+ /*
+ * Search in the list of inherited constraints for any entries on the
+ * same column.
+ */
+ foreach(lc2, old_notnulls)
+ {
+ CookedConstraint *old = (CookedConstraint *) lfirst(lc2);
+
+ if (old->attnum == attnum)
+ {
+ inhcount++;
+ old_notnulls = foreach_delete_current(old_notnulls, lc2);
+ }
+ }
+
+ /*
+ * Determine a constraint name, which may have been specified by the
+ * user, or raise an error if a conflict exists with another
+ * user-specified name.
+ */
+ if (constr->conname)
+ {
+ foreach(lc2, givennames)
+ {
+ if (strcmp(lfirst(lc2), constr->conname) == 0)
+ ereport(ERROR,
+ errcode(ERRCODE_DUPLICATE_OBJECT),
+ errmsg("constraint \"%s\" for relation \"%s\" already exists",
+ constr->conname,
+ RelationGetRelationName(rel)));
+ }
+
+ conname = constr->conname;
+ givennames = lappend(givennames, conname);
+ }
+ else
+ conname = ChooseConstraintName(RelationGetRelationName(rel),
+ get_attname(RelationGetRelid(rel),
+ attnum, false),
+ "not_null",
+ RelationGetNamespace(rel),
+ nnnames);
+ nnnames = lappend(nnnames, conname);
+
+ StoreRelNotNull(rel, conname,
+ attnum, true, is_local,
+ inhcount, constr->is_no_inherit);
+
+ nncols = lappend_int(nncols, attnum);
+ }
+
+ /*
+ * If any column remains in the old_notnulls list, we must create a not-
+ * null constraint marked not-local. Because multiple parents could
+ * specify a not-null constraint for the same column, we must count how
+ * many there are and add to the original inhcount accordingly, deleting
+ * elements we've already processed. We sort the list to make it easy.
+ *
+ * We don't use foreach() here because we have two nested loops over the
+ * constraint list, with possible element deletions in the inner one. If
+ * we used foreach_delete_current() it could only fix up the state of one
+ * of the loops, so it seems cleaner to use looping over list indexes for
+ * both loops. Note that any deletion will happen beyond where the outer
+ * loop is, so its index never needs adjustment.
+ */
+ list_sort(old_notnulls, list_cookedconstr_attnum_cmp);
+ for (int outerpos = 0; outerpos < list_length(old_notnulls); outerpos++)
+ {
+ CookedConstraint *cooked;
+ char *conname = NULL;
+ int add_inhcount = 0;
+ ListCell *lc2;
+
+ cooked = (CookedConstraint *) list_nth(old_notnulls, outerpos);
+ Assert(cooked->contype == CONSTR_NOTNULL);
+
+ /*
+ * Preserve the first non-conflicting constraint name we come across,
+ * if any
+ */
+ if (conname == NULL && cooked->name)
+ conname = cooked->name;
+
+ for (int restpos = outerpos + 1; restpos < list_length(old_notnulls);)
+ {
+ CookedConstraint *other;
+
+ other = (CookedConstraint *) list_nth(old_notnulls, restpos);
+ if (other->attnum == cooked->attnum)
+ {
+ if (conname == NULL && other->name)
+ conname = other->name;
+
+ add_inhcount++;
+ old_notnulls = list_delete_nth_cell(old_notnulls, restpos);
+ }
+ else
+ restpos++;
+ }
+
+ /* If we got a name, make sure it isn't one we've already used */
+ if (conname != NULL)
+ {
+ foreach(lc2, nnnames)
+ {
+ if (strcmp(lfirst(lc2), conname) == 0)
+ {
+ conname = NULL;
+ break;
+ }
+ }
+ }
+
+ /* and choose a name, if needed */
+ if (conname == NULL)
+ conname = ChooseConstraintName(RelationGetRelationName(rel),
+ get_attname(RelationGetRelid(rel),
+ cooked->attnum, false),
+ "not_null",
+ RelationGetNamespace(rel),
+ nnnames);
+ nnnames = lappend(nnnames, conname);
+
+ StoreRelNotNull(rel, conname, cooked->attnum, true,
+ cooked->is_local, cooked->inhcount + add_inhcount,
+ cooked->is_no_inherit);
+
+ nncols = lappend_int(nncols, cooked->attnum);
+ }
+
+ return nncols;
+}
+
/*
* Update the count of constraints in the relation's pg_class tuple.
*
diff --git a/src/backend/catalog/pg_constraint.c b/src/backend/catalog/pg_constraint.c
index 4002317f705..2a725f62809 100644
--- a/src/backend/catalog/pg_constraint.c
+++ b/src/backend/catalog/pg_constraint.c
@@ -21,6 +21,7 @@
#include "access/xact.h"
#include "catalog/catalog.h"
#include "catalog/dependency.h"
+#include "catalog/heap.h"
#include "catalog/indexing.h"
#include "catalog/objectaccess.h"
#include "catalog/pg_constraint.h"
@@ -563,6 +564,291 @@ 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(Oid relid, AttrNumber attnum)
+{
+ Relation pg_constraint;
+ HeapTuple conTup,
+ retval = NULL;
+ SysScanDesc scan;
+ ScanKeyData key;
+
+ pg_constraint = table_open(ConstraintRelationId, AccessShareLock);
+ ScanKeyInit(&key,
+ Anum_pg_constraint_conrelid,
+ BTEqualStrategyNumber, F_OIDEQ,
+ ObjectIdGetDatum(relid));
+ scan = systable_beginscan(pg_constraint, ConstraintRelidTypidNameIndexId,
+ true, NULL, 1, &key);
+
+ while (HeapTupleIsValid(conTup = systable_getnext(scan)))
+ {
+ Form_pg_constraint con = (Form_pg_constraint) GETSTRUCT(conTup);
+ AttrNumber conkey;
+
+ /*
+ * We're looking for a NOTNULL constraint that's marked validated,
+ * with the column we're looking for as the sole element in conkey.
+ */
+ if (con->contype != CONSTRAINT_NOTNULL)
+ continue;
+ if (!con->convalidated)
+ continue;
+
+ conkey = extractNotNullColumn(conTup);
+ if (conkey != attnum)
+ continue;
+
+ /* Found it */
+ retval = heap_copytuple(conTup);
+ break;
+ }
+
+ systable_endscan(scan);
+ table_close(pg_constraint, AccessShareLock);
+
+ return retval;
+}
+
+/*
+ * Find and return the pg_constraint tuple that implements a validated
+ * not-null constraint for the given column of the given relation.
+ */
+HeapTuple
+findNotNullConstraint(Oid relid, const char *colname)
+{
+ AttrNumber attnum = get_attnum(relid, colname);
+
+ return findNotNullConstraintAttnum(relid, attnum);
+}
+
+/*
+ * Given a pg_constraint tuple for a not-null constraint, return the column
+ * number it is for.
+ */
+AttrNumber
+extractNotNullColumn(HeapTuple constrTup)
+{
+ AttrNumber colnum;
+ Datum adatum;
+ ArrayType *arr;
+
+ /* only tuples for not-null constraints should be given */
+ Assert(((Form_pg_constraint) GETSTRUCT(constrTup))->contype == CONSTRAINT_NOTNULL);
+
+ adatum = SysCacheGetAttrNotNull(CONSTROID, constrTup,
+ Anum_pg_constraint_conkey);
+ arr = DatumGetArrayTypeP(adatum); /* ensure not toasted */
+ if (ARR_NDIM(arr) != 1 ||
+ ARR_HASNULL(arr) ||
+ ARR_ELEMTYPE(arr) != INT2OID ||
+ ARR_DIMS(arr)[0] != 1)
+ elog(ERROR, "conkey is not a 1-D smallint array");
+
+ memcpy(&colnum, ARR_DATA_PTR(arr), sizeof(AttrNumber));
+
+ if ((Pointer) arr != DatumGetPointer(adatum))
+ pfree(arr); /* free de-toasted copy, if any */
+
+ return colnum;
+}
+
+/*
+ * AdjustNotNullInheritance1
+ * Adjust inheritance count for a single not-null constraint
+ *
+ * Adjust inheritance count, and possibly islocal status, for the not-null
+ * constraint row of the given column, if it exists, and return true.
+ * If no not-null constraint is found for the column, return false.
+ */
+bool
+AdjustNotNullInheritance1(Oid relid, AttrNumber attnum, int count)
+{
+ HeapTuple tup;
+
+ tup = findNotNullConstraintAttnum(relid, attnum);
+ if (HeapTupleIsValid(tup))
+ {
+ Relation pg_constraint;
+ Form_pg_constraint conform;
+
+ pg_constraint = table_open(ConstraintRelationId, RowExclusiveLock);
+ conform = (Form_pg_constraint) GETSTRUCT(tup);
+ if (count > 0)
+ conform->coninhcount += count;
+
+ /* sanity check */
+ if (conform->coninhcount < 0)
+ elog(ERROR, "invalid inhcount %d for constraint \"%s\" on relation \"%s\"",
+ conform->coninhcount, NameStr(conform->conname),
+ get_rel_name(relid));
+
+ /*
+ * If the constraints are no longer inherited, mark them local. It's
+ * arguable that we should drop them instead, but it's hard to see
+ * that being better. The user can drop it manually later.
+ */
+ if (conform->coninhcount == 0)
+ conform->conislocal = true;
+
+ CatalogTupleUpdate(pg_constraint, &tup->t_self, tup);
+
+ table_close(pg_constraint, RowExclusiveLock);
+
+ return true;
+ }
+
+ return false;
+}
+
+/*
+ * AdjustNotNullInheritance
+ * Adjust not-null constraints' inhcount/islocal for
+ * ALTER TABLE [NO] INHERITS
+ *
+ * Mark the NOT NULL constraints for the given relation columns as
+ * inherited, so that they can't be dropped.
+ *
+ * Caller must have checked beforehand that attnotnull was set for all
+ * columns. However, some of those could be set because of a primary
+ * key, so throw a proper user-visible error if one is not found.
+ */
+void
+AdjustNotNullInheritance(Oid relid, Bitmapset *columns, int count)
+{
+ Relation pg_constraint;
+ int attnum;
+
+ pg_constraint = table_open(ConstraintRelationId, RowExclusiveLock);
+
+ /*
+ * Scan the set of columns and bump inhcount for each.
+ */
+ attnum = -1;
+ while ((attnum = bms_next_member(columns, attnum)) >= 0)
+ {
+ HeapTuple tup;
+ Form_pg_constraint conform;
+
+ tup = findNotNullConstraintAttnum(relid, attnum);
+ if (!HeapTupleIsValid(tup))
+ ereport(ERROR,
+ errcode(ERRCODE_DATATYPE_MISMATCH),
+ errmsg("column \"%s\" in child table must be marked NOT NULL",
+ get_attname(relid, attnum,
+ false)));
+
+ conform = (Form_pg_constraint) GETSTRUCT(tup);
+ conform->coninhcount += count;
+ if (conform->coninhcount < 0)
+ elog(ERROR, "invalid inhcount %d for constraint \"%s\" on relation \"%s\"",
+ conform->coninhcount, NameStr(conform->conname),
+ get_rel_name(relid));
+
+ /*
+ * If the constraints are no longer inherited, mark them local. It's
+ * arguable that we should drop them instead, but it's hard to see
+ * that being better. The user can drop it manually later.
+ */
+ if (conform->coninhcount == 0)
+ conform->conislocal = true;
+
+ CatalogTupleUpdate(pg_constraint, &tup->t_self, tup);
+ }
+
+ table_close(pg_constraint, RowExclusiveLock);
+}
+
+/*
+ * RelationGetNotNullConstraints
+ * Return the list of not-null constraints for the given rel
+ *
+ * Caller can request cooked constraints, or raw.
+ *
+ * This is seldom needed, so we just scan pg_constraint each time.
+ *
+ * XXX This is only used to create derived tables, so NO INHERIT constraints
+ * are always skipped.
+ */
+List *
+RelationGetNotNullConstraints(Oid relid, bool cooked)
+{
+ List *notnulls = NIL;
+ Relation constrRel;
+ HeapTuple htup;
+ SysScanDesc conscan;
+ ScanKeyData skey;
+
+ constrRel = table_open(ConstraintRelationId, AccessShareLock);
+ ScanKeyInit(&skey,
+ Anum_pg_constraint_conrelid,
+ BTEqualStrategyNumber, F_OIDEQ,
+ ObjectIdGetDatum(relid));
+ conscan = systable_beginscan(constrRel, ConstraintRelidTypidNameIndexId, true,
+ NULL, 1, &skey);
+
+ while (HeapTupleIsValid(htup = systable_getnext(conscan)))
+ {
+ Form_pg_constraint conForm = (Form_pg_constraint) GETSTRUCT(htup);
+ AttrNumber colnum;
+
+ if (conForm->contype != CONSTRAINT_NOTNULL)
+ continue;
+ if (conForm->connoinherit)
+ continue;
+
+ colnum = extractNotNullColumn(htup);
+
+ if (cooked)
+ {
+ CookedConstraint *cooked;
+
+ cooked = (CookedConstraint *) palloc(sizeof(CookedConstraint));
+
+ cooked->contype = CONSTR_NOTNULL;
+ cooked->name = pstrdup(NameStr(conForm->conname));
+ cooked->attnum = colnum;
+ cooked->expr = NULL;
+ cooked->skip_validation = false;
+ cooked->is_local = true;
+ cooked->inhcount = 0;
+ cooked->is_no_inherit = conForm->connoinherit;
+
+ notnulls = lappend(notnulls, cooked);
+ }
+ else
+ {
+ Constraint *constr;
+
+ constr = makeNode(Constraint);
+ constr->contype = CONSTR_NOTNULL;
+ constr->conname = pstrdup(NameStr(conForm->conname));
+ constr->deferrable = false;
+ constr->initdeferred = false;
+ constr->location = -1;
+ constr->keys = list_make1(makeString(get_attname(relid, colnum,
+ false)));
+ constr->skip_validation = false;
+ constr->initially_valid = true;
+ notnulls = lappend(notnulls, constr);
+ }
+ }
+
+ systable_endscan(conscan);
+ table_close(constrRel, AccessShareLock);
+
+ return notnulls;
+}
+
+
+/*
* Delete a single constraint record.
*/
void
diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c
index f77de4e7c99..47c900445c7 100644
--- a/src/backend/commands/tablecmds.c
+++ b/src/backend/commands/tablecmds.c
@@ -200,7 +200,7 @@ typedef struct AlteredTableInfo
} AlteredTableInfo;
/* Struct describing one new constraint to check in Phase 3 scan */
-/* Note: new NOT NULL constraints are handled elsewhere */
+/* Note: new not-null constraints are handled elsewhere */
typedef struct NewConstraint
{
char *name; /* Constraint name, or NULL if none */
@@ -351,7 +351,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 **supnotnulls);
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);
@@ -432,16 +433,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 void ATExecCheckNotNull(AlteredTableInfo *tab, Relation rel,
- const char *colName, LOCKMODE lockmode);
+static ObjectAddress ATExecDropNotNull(Relation rel, const char *colName, bool recurse,
+ LOCKMODE lockmode);
+static bool set_attnotnull(List **wqueue, Relation rel,
+ AttrNumber attnum, bool recurse, LOCKMODE lockmode);
+static ObjectAddress ATExecSetNotNull(List **wqueue, Relation rel,
+ char *constrname, char *colName,
+ bool recurse, bool recursing,
+ List **readyRels, LOCKMODE lockmode);
+static ObjectAddress ATExecSetAttNotNull(List **wqueue, Relation rel,
+ const char *colName, LOCKMODE lockmode);
static bool NotNullImpliedByRelConstraints(Relation rel, Form_pg_attribute attr);
static bool ConstraintImpliedByRelConstraint(Relation scanrel,
List *testConstraint, List *provenConstraint);
@@ -470,6 +471,8 @@ static ObjectAddress ATExecDropColumn(List **wqueue, Relation rel, const char *c
bool recurse, bool recursing,
bool missing_ok, LOCKMODE lockmode,
ObjectAddresses *addrs);
+static void ATPrepAddPrimaryKey(List **wqueue, Relation rel, AlterTableCmd *cmd,
+ LOCKMODE lockmode, AlterTableUtilityContext *context);
static ObjectAddress ATExecAddIndex(AlteredTableInfo *tab, Relation rel,
IndexStmt *stmt, bool is_rebuild, LOCKMODE lockmode);
static ObjectAddress ATExecAddStatistics(AlteredTableInfo *tab, Relation rel,
@@ -481,11 +484,11 @@ static ObjectAddress ATExecAddConstraint(List **wqueue,
static char *ChooseForeignKeyConstraintNameAddition(List *colnames);
static ObjectAddress ATExecAddIndexConstraint(AlteredTableInfo *tab, Relation rel,
IndexStmt *stmt, LOCKMODE lockmode);
-static ObjectAddress ATAddCheckConstraint(List **wqueue,
- AlteredTableInfo *tab, Relation rel,
- Constraint *constr,
- bool recurse, bool recursing, bool is_readd,
- LOCKMODE lockmode);
+static ObjectAddress ATAddCheckNNConstraint(List **wqueue,
+ AlteredTableInfo *tab, Relation rel,
+ Constraint *constr,
+ bool recurse, bool recursing, bool is_readd,
+ LOCKMODE lockmode);
static ObjectAddress ATAddForeignKeyConstraint(List **wqueue, AlteredTableInfo *tab,
Relation rel, Constraint *fkconstraint,
bool recurse, bool recursing,
@@ -542,6 +545,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,
@@ -614,10 +622,12 @@ static void ComputePartitionAttrs(ParseState *pstate, Relation rel, List *partPa
static void CreateInheritance(Relation child_rel, Relation parent_rel);
static void RemoveInheritance(Relation child_rel, Relation parent_rel,
bool expect_detached);
+static void ATInheritAdjustNotNulls(Relation parent_rel, Relation child_rel,
+ int inhcount);
static ObjectAddress ATExecAttachPartition(List **wqueue, Relation rel,
PartitionCmd *cmd,
AlterTableUtilityContext *context);
-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);
@@ -635,6 +645,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);
@@ -672,8 +683,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;
@@ -863,12 +876,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);
@@ -1251,6 +1265,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 declared directly with
+ * those that come from parent relations (making sure to count inheritance
+ * appropriately for each), create them, and set the attnotnull flag on
+ * columns that don't yet have it.
+ */
+ nncols = AddRelationNotNullConstraints(rel, stmt->nnconstraints,
+ old_notnulls);
+ foreach(listptr, nncols)
+ set_attnotnull(NULL, rel, lfirst_int(listptr), false, NoLock);
+
ObjectAddressSet(address, RelationRelationId, relationId);
/*
@@ -2299,6 +2324,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.
+ * 'supnotnulls' receives a list of CookedConstraints that corresponds to
+ * constraints coming from inheritance parents.
*
* Return value:
* Completed schema list.
@@ -2327,9 +2354,12 @@ storage_name(char c)
* If the same attribute name appears multiple times, then it appears
* in the result table in the proper location for its first appearance.
*
- * Constraints (including NOT NULL constraints) for the child table
+ * 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.
@@ -2348,10 +2378,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 +2493,12 @@ 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 +2587,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(RelationGetRelid(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 +2696,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 not-null constraint. Partitioning doesn't
+ * need this, because the PK itself is going to be cloned to
+ * the partition.
*/
- def->is_not_null |= attribute->attnotnull;
+ 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
+ */
+ 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 +2761,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 +2782,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 +2953,23 @@ 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);
+
+ Assert(nn->contype == CONSTR_NOTNULL);
+
+ nn->attnum = newattmap->attnums[nn->attnum - 1];
+ nn->is_local = false;
+ nn->inhcount = 1;
+
+ nnconstraints = lappend(nnconstraints, nn);
+ }
+
free_attrmap(newattmap);
/*
@@ -2972,7 +3097,7 @@ MergeAttributes(List *schema, List *supers, char relpersistence,
}
/*
- * Merge of NOT NULL constraints = OR 'em together
+ * Merge of not-null constraints = OR 'em together
*/
def->is_not_null |= newdef->is_not_null;
@@ -3051,8 +3176,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. Also, merge column defaults.
*/
if (is_partition)
{
@@ -3069,7 +3193,6 @@ MergeAttributes(List *schema, List *supers, char relpersistence,
if (strcmp(coldef->colname, restdef->colname) == 0)
{
found = true;
- coldef->is_not_null |= restdef->is_not_null;
/*
* Check for conflicts related to generated columns.
@@ -3158,6 +3281,8 @@ MergeAttributes(List *schema, List *supers, char relpersistence,
}
*supconstr = constraints;
+ *supnotnulls = nnconstraints;
+
return schema;
}
@@ -3769,7 +3894,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 +4482,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:
@@ -4492,15 +4621,6 @@ AlterTableGetLockLevel(List *cmds)
cmd_lockmode = ShareUpdateExclusiveLock;
break;
- case AT_CheckNotNull:
-
- /*
- * This only examines the table's schema; but lock must be
- * strong enough to prevent concurrent DROP NOT NULL.
- */
- cmd_lockmode = AccessShareLock;
- break;
-
default: /* oops */
elog(ERROR, "unrecognized alter table type: %d",
(int) cmd->subtype);
@@ -4652,21 +4772,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);
+ /* Set up recursion for phase 2; no other prep needed */
+ if (recurse)
+ cmd->recurse = true;
pass = AT_PASS_COL_ATTRS;
break;
- case AT_CheckNotNull: /* check column is already marked NOT NULL */
+ 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);
- /* No command-specific prep needed */
pass = AT_PASS_COL_ATTRS;
break;
case AT_DropExpression: /* ALTER COLUMN DROP EXPRESSION */
@@ -5045,13 +5167,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_CheckNotNull: /* check column is already marked NOT NULL */
- ATExecCheckNotNull(tab, rel, cmd->name, lockmode);
+ case AT_SetAttNotNull: /* set pg_attribute.attnotnull */
+ address = ATExecSetAttNotNull(wqueue, rel, cmd->name, lockmode);
break;
case AT_DropExpression:
address = ATExecDropExpression(rel, cmd->name, cmd->missing_ok, lockmode);
@@ -5387,21 +5510,23 @@ 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:
- /* This command never recurses */
- /* No command-specific prep needed */
+
+ /*
+ * A primary key on a inheritance parent needs supporting NOT
+ * NULL constraint on its children; enqueue commands to create
+ * those or mark them inherited if they already exist.
+ */
+ ATPrepAddPrimaryKey(wqueue, rel, cmd2, lockmode, context);
pass = AT_PASS_ADD_INDEX;
break;
case AT_AddIndexConstraint:
- /* This command never recurses */
- /* No command-specific prep needed */
+ /* as above */
+ ATPrepAddPrimaryKey(wqueue, rel, cmd2, lockmode, context);
pass = AT_PASS_ADD_INDEXCONSTR;
break;
case AT_AddConstraint:
@@ -5845,7 +5970,7 @@ ATRewriteTable(AlteredTableInfo *tab, Oid OIDNewHeap, LOCKMODE lockmode)
{
/*
* If we are rebuilding the tuples OR if we added any new but not
- * verified NOT NULL constraints, check all not-null constraints. This
+ * verified not-null constraints, check all not-null constraints. This
* is a bit of overkill but it minimizes risk of bugs, and
* heap_attisnull is a pretty cheap test anyway.
*/
@@ -6067,6 +6192,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,10 +6301,10 @@ 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:
- return NULL; /* not real grammar */
case AT_SetStatistics:
return "ALTER COLUMN ... SET STATISTICS";
case AT_SetOptions:
@@ -6774,8 +6900,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)
{
@@ -7044,7 +7169,7 @@ ATExecAddColumn(List **wqueue, AlteredTableInfo *tab, Relation rel,
* the effect of NULL values in the new column.
*
* An exception occurs when the new column is of a domain type: the domain
- * might have a NOT NULL constraint, or a check constraint that indirectly
+ * might have a not-null constraint, or a check constraint that indirectly
* rejects nulls. If there are any domain constraints then we construct
* an explicit NULL default value that will be passed through
* CoerceToDomain processing. (This is a tad inefficient, since it causes
@@ -7290,42 +7415,21 @@ 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;
+ List *readyRels;
/*
* lookup the attribute
@@ -7340,6 +7444,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,62 +7468,37 @@ 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);
@@ -7428,19 +7516,35 @@ 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(RelationGetRelid(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, "could not find not-null constraint on column \"%s\", relation \"%s\"",
+ colName, RelationGetRelationName(rel));
}
- else
- address = InvalidObjectAddress;
+
+ readyRels = NIL;
+ dropconstraint_internal(rel, conTup, DROP_RESTRICT, recurse, false,
+ false, &readyRels, lockmode);
+
+ heap_freetuple(conTup);
InvokeObjectPostAlterHook(RelationRelationId,
RelationGetRelid(rel), attnum);
@@ -7451,102 +7555,137 @@ 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.
+ *
+ * Returns true if the flag was set in any table, otherwise false.
*/
-
-static void
-ATPrepSetNotNull(List **wqueue, Relation rel,
- AlterTableCmd *cmd, bool recurse, bool recursing,
- LOCKMODE lockmode, AlterTableUtilityContext *context)
+static bool
+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;
+ bool retval = false;
- /*
- * 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;
+ }
+
+ retval = 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 (recurse)
{
- AlterTableCmd *newcmd = makeNode(AlterTableCmd);
+ List *children;
+ ListCell *lc;
+
+ children = find_inheritance_children(RelationGetRelid(rel), lockmode);
+ foreach(lc, children)
+ {
+ Oid childrelid = lfirst_oid(lc);
+ Relation childrel;
+ AttrNumber childattno;
+
+ /* find_inheritance_children already got lock */
+ childrel = table_open(childrelid, NoLock);
+ CheckTableNotInUse(childrel, "ALTER TABLE");
- 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));
+ retval |= set_attnotnull(wqueue, childrel, childattno,
+ recurse, lockmode);
+ table_close(childrel, NoLock);
+ }
}
- else
- ATSimpleRecursion(wqueue, rel, cmd, recurse, lockmode, context);
+
+ return retval;
}
/*
- * 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;
+ bool is_no_inherit = false;
+ 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)
+ {
+ Assert(!recursing);
+ 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,80 +7693,178 @@ 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, &copytup->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;
+
+ systable_endscan(conscan);
+ table_close(constr_rel, RowExclusiveLock);
+
+ /*
+ * If we're asked not to recurse, and children exist, raise an error for
+ * partitioned tables. For inheritance, we act as if NO INHERIT had been
+ * specified.
+ */
+ if (!recurse &&
+ find_inheritance_children(RelationGetRelid(rel),
+ NoLock) != NIL)
+ {
+ if (rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
+ ereport(ERROR,
+ errcode(ERRCODE_INVALID_TABLE_DEFINITION),
+ errmsg("constraint must be added to child tables too"),
+ errhint("Do not specify the ONLY keyword."));
+ else
+ is_no_inherit = true;
+ }
+
+ /*
+ * No constraint exists; we must add one. First determine a name to use,
+ * if we haven't already.
+ */
+ if (!recursing)
+ {
+ Assert(conName == NULL);
+ conName = ChooseConstraintName(RelationGetRelationName(rel),
+ colName, "not_null",
+ RelationGetNamespace(rel),
+ NIL);
+ }
+ constraint = makeNode(Constraint);
+ constraint->contype = CONSTR_NOTNULL;
+ constraint->conname = conName;
+ constraint->deferrable = false;
+ constraint->initdeferred = false;
+ constraint->location = -1;
+ constraint->keys = list_make1(makeString(colName));
+ constraint->is_no_inherit = is_no_inherit;
+ constraint->inhcount = recursing ? 1 : 0;
+ constraint->skip_validation = false;
+ constraint->initially_valid = true;
+
+ /* and do it */
+ cooked = AddRelationNewConstraints(rel, NIL, list_make1(constraint),
+ false, !recursing, false, NULL);
+ ccon = linitial(cooked);
+ ObjectAddressSet(address, ConstraintRelationId, ccon->conoid);
InvokeObjectPostAlterHook(RelationRelationId,
RelationGetRelid(rel), attnum);
- table_close(attr_rel, RowExclusiveLock);
+ /*
+ * 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 CHECK NOT NULL
+ * ALTER TABLE ALTER COLUMN SET ATTNOTNULL
*
- * This doesn't exist in the grammar, but we generate AT_CheckNotNull
- * commands against the partitions of a partitioned table if the user
- * writes ALTER TABLE ONLY ... SET NOT NULL on the partitioned table,
- * or tries to create a primary key on it (which internally creates
- * AT_SetNotNull on the partitioned table). Such a command doesn't
- * allow us to actually modify any partition, but we want to let it
- * go through if the partitions are already properly marked.
- *
- * In future, this might need to adjust the child table's state, likely
- * by incrementing an inheritance count for the attnotnull constraint.
- * For now we need only check for the presence of the flag.
+ * 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
-ATExecCheckNotNull(AlteredTableInfo *tab, Relation rel,
- const char *colName, LOCKMODE lockmode)
+static ObjectAddress
+ATExecSetAttNotNull(List **wqueue, Relation rel,
+ const char *colName, LOCKMODE lockmode)
{
- HeapTuple tuple;
-
- tuple = SearchSysCacheAttName(RelationGetRelid(rel), colName);
+ AttrNumber attnum;
+ ObjectAddress address = InvalidObjectAddress;
- 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))));
+ errcode(ERRCODE_UNDEFINED_COLUMN),
+ errmsg("column \"%s\" of relation \"%s\" does not exist",
+ colName, RelationGetRelationName(rel)));
- if (!((Form_pg_attribute) GETSTRUCT(tuple))->attnotnull)
- ereport(ERROR,
- (errcode(ERRCODE_INVALID_TABLE_DEFINITION),
- errmsg("constraint must be added to child tables too"),
- errdetail("Column \"%s\" of relation \"%s\" is not already NOT NULL.",
- colName, RelationGetRelationName(rel)),
- errhint("Do not specify the ONLY keyword.")));
+ /*
+ * Make the change, if necessary, and only if so report the column as
+ * changed
+ */
+ if (set_attnotnull(wqueue, rel, attnum, false, lockmode))
+ ObjectAddressSubSet(address, RelationRelationId,
+ RelationGetRelid(rel), attnum);
- ReleaseSysCache(tuple);
+ return address;
}
/*
@@ -8678,6 +8915,71 @@ ATExecDropColumn(List **wqueue, Relation rel, const char *colName,
}
/*
+ * Prepare to add a primary key on an inheritance parent, by adding NOT NULL
+ * constraint on its children.
+ */
+static void
+ATPrepAddPrimaryKey(List **wqueue, Relation rel, AlterTableCmd *cmd,
+ LOCKMODE lockmode, AlterTableUtilityContext *context)
+{
+ List *children;
+ List *newconstrs = NIL;
+ ListCell *lc;
+ IndexStmt *stmt;
+
+ /* No work if no legacy inheritance children are present */
+ if (rel->rd_rel->relkind != RELKIND_RELATION ||
+ !rel->rd_rel->relhassubclass)
+ return;
+
+ children = find_inheritance_children(RelationGetRelid(rel), lockmode);
+
+ stmt = castNode(IndexStmt, cmd->def);
+ foreach(lc, stmt->indexParams)
+ {
+ IndexElem *elem = lfirst_node(IndexElem, lc);
+ Constraint *nnconstr;
+
+ Assert(elem->expr == NULL);
+
+ nnconstr = makeNode(Constraint);
+ nnconstr->contype = CONSTR_NOTNULL;
+ nnconstr->conname = NULL; /* XXX use PK name? */
+ nnconstr->inhcount = 1;
+ nnconstr->deferrable = false;
+ nnconstr->initdeferred = false;
+ nnconstr->location = -1;
+ nnconstr->keys = list_make1(makeString(elem->name));
+ nnconstr->skip_validation = false;
+ nnconstr->initially_valid = true;
+
+ newconstrs = lappend(newconstrs, nnconstr);
+ }
+
+ foreach(lc, children)
+ {
+ Oid childrelid = lfirst_oid(lc);
+ Relation childrel = table_open(childrelid, NoLock);
+ AlterTableCmd *newcmd = makeNode(AlterTableCmd);
+ ListCell *lc2;
+
+ newcmd->subtype = AT_AddConstraint;
+ newcmd->recurse = true;
+
+ foreach(lc2, newconstrs)
+ {
+ /* ATPrepCmd copies newcmd, so we can scribble on it here */
+ newcmd->def = lfirst(lc2);
+
+ ATPrepCmd(wqueue, childrel, newcmd,
+ true, false, lockmode, context);
+ }
+
+ table_close(childrel, NoLock);
+ }
+}
+
+/*
* ALTER TABLE ADD INDEX
*
* There is no such command in the grammar, but parse_utilcmd.c converts
@@ -8872,17 +9174,18 @@ 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,
- lockmode);
+ ATAddCheckNNConstraint(wqueue, tab, rel,
+ newConstraint, recurse, false, is_readd,
+ lockmode);
break;
case CONSTR_FOREIGN:
@@ -8963,9 +9266,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.
*
@@ -8978,9 +9281,9 @@ ChooseForeignKeyConstraintNameAddition(List *colnames)
* the parent table and pass that down.
*/
static ObjectAddress
-ATAddCheckConstraint(List **wqueue, AlteredTableInfo *tab, Relation rel,
- Constraint *constr, bool recurse, bool recursing,
- bool is_readd, LOCKMODE lockmode)
+ATAddCheckNNConstraint(List **wqueue, AlteredTableInfo *tab, Relation rel,
+ Constraint *constr, bool recurse, bool recursing,
+ bool is_readd, LOCKMODE lockmode)
{
List *newcons;
ListCell *lcon;
@@ -9018,7 +9321,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,11 +9337,19 @@ 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);
}
/* At this point we must have a locked-down name to use */
- Assert(constr->conname != NULL);
+ Assert(newcons == NIL || constr->conname != NULL);
/* Advance command counter in case same table is visited multiple times */
CommandCounterIncrement();
@@ -9076,6 +9387,12 @@ ATAddCheckConstraint(List **wqueue, AlteredTableInfo *tab, Relation rel,
(errcode(ERRCODE_INVALID_TABLE_DEFINITION),
errmsg("constraint must be added to child tables too")));
+ /*
+ * The constraint must appear as inherited in children, so create a
+ * modified constraint object to use.
+ */
+ constr = copyObject(constr);
+ constr->inhcount = 1;
foreach(child, children)
{
Oid childrelid = lfirst_oid(child);
@@ -9089,9 +9406,13 @@ ATAddCheckConstraint(List **wqueue, AlteredTableInfo *tab, Relation rel,
/* Find or create work queue entry for this table */
childtab = ATGetQueueEntry(wqueue, childrel);
- /* Recurse to child */
- ATAddCheckConstraint(wqueue, childtab, childrel,
- constr, recurse, true, is_readd, lockmode);
+ /*
+ * Recurse to child. XXX if we didn't create a constraint on the
+ * parent because it already existed, and we do create one on a child,
+ * should we return that child's constraint ObjectAddress here?
+ */
+ ATAddCheckNNConstraint(wqueue, childtab, childrel,
+ constr, recurse, true, is_readd, lockmode);
table_close(childrel, NoLock);
}
@@ -11958,16 +12279,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 +12312,238 @@ ATExecDropConstraint(Relation rel, const char *constrName,
/* There can be at most one matching row */
if (HeapTupleIsValid(tuple = systable_getnext(scan)))
{
- ObjectAddress conobj;
+ List *readyRels = NIL;
+
+ dropconstraint_internal(rel, tuple, behavior, recurse, recursing,
+ missing_ok, &readyRels, 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;
- /* Must match lock taken by RemoveTriggerById: */
- frel = table_open(con->confrelid, AccessExclusiveLock);
- CheckTableNotInUse(frel, "ALTER TABLE");
- table_close(frel, NoLock);
- }
+ if (list_member_oid(*readyRels, RelationGetRelid(rel)))
+ return InvalidObjectAddress;
+ *readyRels = lappend_oid(*readyRels, RelationGetRelid(rel));
- /*
- * Perform the actual constraint deletion
- */
- conobj.classId = ConstraintRelationId;
- conobj.objectId = con->oid;
- conobj.objectSubId = 0;
+ conrel = table_open(ConstraintRelationId, RowExclusiveLock);
- performDeletion(&conobj, behavior, 0);
+ con = (Form_pg_constraint) GETSTRUCT(constraintTup);
+ constrName = NameStr(con->conname);
- found = true;
+ /* Don't allow drop of inherited constraints */
+ if (con->coninhcount > 0 && !recursing)
+ ereport(ERROR,
+ (errcode(ERRCODE_INVALID_TABLE_DEFINITION),
+ errmsg("cannot drop inherited constraint \"%s\" of relation \"%s\"",
+ constrName, RelationGetRelationName(rel))));
+
+ /*
+ * See if we have a not-null constraint or a PRIMARY KEY. If so, we have
+ * more checks and actions below, so obtain the list of columns that are
+ * constrained by the constraint being dropped.
+ */
+ if (con->contype == CONSTRAINT_NOTNULL)
+ {
+ AttrNumber colnum = 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;
- systable_endscan(scan);
+ dropping_pk = true;
- if (!found)
+ 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))
{
- if (!missing_ok)
- {
- ereport(ERROR,
- (errcode(ERRCODE_UNDEFINED_OBJECT),
- errmsg("constraint \"%s\" of relation \"%s\" does not exist",
- constrName, RelationGetRelationName(rel))));
- }
- else
+ 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 attribute %d of relation %u",
+ attnum, RelationGetRelid(rel));
+ 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(RelationGetRelid(rel), attnum);
+ if (contup ||
+ bms_is_member(attnum - FirstLowInvalidHeapAttributeNumber,
+ pkcols))
+ continue;
+
+ /*
+ * It's not valid to drop the not-null constraint for a GENERATED
+ * AS IDENTITY column.
+ */
+ if (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 not-null constraint for a column in
+ * the replica identity index, either. (FULL is not affected.)
+ */
+ 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.
+ * For partitioned tables, non-CHECK, non-NOT-NULL inherited constraints
+ * are dropped via the dependency mechanism, so we're done here.
*/
- if (contype != CONSTRAINT_CHECK &&
+ if (con->contype != CONSTRAINT_CHECK &&
+ con->contype != CONSTRAINT_NOTNULL &&
rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
{
table_close(conrel, RowExclusiveLock);
- return;
+ return conobj;
}
/*
@@ -12094,52 +12568,79 @@ 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 copy_tuple;
+ HeapTuple tuple;
+ Form_pg_constraint childcon;
+
+ 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))));
-
- copy_tuple = heap_copytuple(tuple);
-
- systable_endscan(scan);
+ /*
+ * We search for not-null constraint by column number, and other
+ * constraints by name.
+ */
+ if (con->contype == CONSTRAINT_NOTNULL)
+ {
+ tuple = findNotNullConstraint(childrelid, colname);
+ if (!HeapTupleIsValid(tuple))
+ elog(ERROR, "cache lookup failed for not-null constraint on column \"%s\" of relation %u",
+ colname, RelationGetRelid(childrel));
+ }
+ else
+ {
+ SysScanDesc scan;
+ ScanKeyData skey[3];
+
+ ScanKeyInit(&skey[0],
+ Anum_pg_constraint_conrelid,
+ BTEqualStrategyNumber, F_OIDEQ,
+ ObjectIdGetDatum(childrelid));
+ ScanKeyInit(&skey[1],
+ Anum_pg_constraint_contypid,
+ BTEqualStrategyNumber, F_OIDEQ,
+ ObjectIdGetDatum(InvalidOid));
+ ScanKeyInit(&skey[2],
+ Anum_pg_constraint_conname,
+ BTEqualStrategyNumber, F_NAMEEQ,
+ CStringGetDatum(constrName));
+ scan = systable_beginscan(conrel, ConstraintRelidTypidNameIndexId,
+ true, NULL, 3, skey);
+ /* There can only be one, so no need to loop */
+ tuple = systable_getnext(scan);
+ if (!HeapTupleIsValid(tuple))
+ ereport(ERROR,
+ (errcode(ERRCODE_UNDEFINED_OBJECT),
+ errmsg("constraint \"%s\" of relation \"%s\" does not exist",
+ constrName,
+ RelationGetRelationName(childrel))));
+ tuple = heap_copytuple(tuple);
+ systable_endscan(scan);
+ }
- con = (Form_pg_constraint) GETSTRUCT(copy_tuple);
+ childcon = (Form_pg_constraint) GETSTRUCT(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);
+ childrelid, NameStr(childcon->conname));
if (recurse)
{
@@ -12147,18 +12648,18 @@ 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, tuple, behavior,
+ recurse, true, missing_ok, readyRels,
+ lockmode);
}
else
{
/* Child constraint must survive my deletion */
- con->coninhcount--;
- CatalogTupleUpdate(conrel, &copy_tuple->t_self, copy_tuple);
+ childcon->coninhcount--;
+ CatalogTupleUpdate(conrel, &tuple->t_self, tuple);
/* Make update visible */
CommandCounterIncrement();
@@ -12167,25 +12668,94 @@ ATExecDropConstraint(Relation rel, const char *constrName,
else
{
/*
- * If we were told to drop ONLY in this table (no recursion), we
- * need to mark the inheritors' constraints as locally defined
- * rather than inherited.
+ * If we were told to drop ONLY in this table (no recursion) and
+ * there are no further parents for this constraint, we need to
+ * mark the inheritors' constraints as locally defined rather than
+ * inherited.
*/
- con->coninhcount--;
- con->conislocal = true;
+ childcon->coninhcount--;
+ if (childcon->coninhcount == 0)
+ childcon->conislocal = true;
- CatalogTupleUpdate(conrel, &copy_tuple->t_self, copy_tuple);
+ CatalogTupleUpdate(conrel, &tuple->t_self, tuple);
/* Make update visible */
CommandCounterIncrement();
}
- heap_freetuple(copy_tuple);
+ heap_freetuple(tuple);
table_close(childrel, NoLock);
}
+ /*
+ * In addition, when dropping a primary key from a legacy-inheritance
+ * parent table, we must recurse to children to mark the corresponding NOT
+ * NULL constraint as no longer inherited, or drop it if this its last
+ * reference.
+ */
+ if (con->contype == CONSTRAINT_PRIMARY &&
+ rel->rd_rel->relkind == RELKIND_RELATION &&
+ rel->rd_rel->relhassubclass)
+ {
+ List *colnames = NIL;
+ ListCell *lc;
+ List *pkready = NIL;
+
+ /*
+ * Because primary keys are always marked as NO INHERIT, we don't have
+ * a list of children yet, so obtain one now.
+ */
+ children = find_inheritance_children(RelationGetRelid(rel), lockmode);
+
+ /*
+ * Find out the list of column names to process. Fortunately, we
+ * already have the list of column numbers.
+ */
+ foreach(lc, unconstrained_cols)
+ {
+ colnames = lappend(colnames, get_attname(RelationGetRelid(rel),
+ lfirst_int(lc), false));
+ }
+
+ foreach(child, children)
+ {
+ Oid childrelid = lfirst_oid(child);
+ Relation childrel;
+
+ if (list_member_oid(pkready, childrelid))
+ continue; /* child already processed */
+
+ /* find_inheritance_children already got lock */
+ childrel = table_open(childrelid, NoLock);
+ CheckTableNotInUse(childrel, "ALTER TABLE");
+
+ foreach(lc, colnames)
+ {
+ HeapTuple contup;
+ char *colName = lfirst(lc);
+
+ contup = findNotNullConstraint(childrelid, colName);
+ if (contup == NULL)
+ elog(ERROR, "cache lookup failed for not-null constraint on column \"%s\", relation \"%s\"",
+ colName, RelationGetRelationName(childrel));
+
+ dropconstraint_internal(childrel, contup,
+ DROP_RESTRICT, true, true,
+ false, &pkready,
+ lockmode);
+ pkready = NIL;
+ }
+
+ table_close(childrel, NoLock);
+
+ pkready = lappend_oid(pkready, childrelid);
+ }
+ }
+
table_close(conrel, RowExclusiveLock);
+
+ return conobj;
}
/*
@@ -13262,9 +13832,10 @@ ATPostAlterTypeCleanup(List **wqueue, AlteredTableInfo *tab, LOCKMODE lockmode)
/*
* If the constraint is inherited (only), we don't want to inject a
- * new definition here; it'll get recreated when ATAddCheckConstraint
- * recurses from adding the parent table's constraint. But we had to
- * carry the info this far so that we can drop the constraint below.
+ * new definition here; it'll get recreated when
+ * ATAddCheckNNConstraint recurses from adding the parent table's
+ * constraint. But we had to carry the info this far so that we can
+ * drop the constraint below.
*/
if (!conislocal)
continue;
@@ -13511,10 +14082,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
@@ -14988,6 +15559,13 @@ ATExecAddInherit(Relation child_rel, RangeVar *parent, LOCKMODE lockmode)
/* OK to create inheritance */
CreateInheritance(child_rel, parent_rel);
+ /*
+ * If parent_rel has a primary key, then child_rel has not-null
+ * constraints that make these columns as non nullable. Make those
+ * constraints as inherited.
+ */
+ ATInheritAdjustNotNulls(parent_rel, child_rel, 1);
+
ObjectAddressSet(address, RelationRelationId,
RelationGetRelid(parent_rel));
@@ -15181,13 +15759,21 @@ MergeAttributesIntoExisting(Relation child_rel, Relation parent_rel)
/*
* Check child doesn't discard NOT NULL property. (Other
- * constraints are checked elsewhere.)
+ * constraints are checked elsewhere.) However, if the constraint
+ * is NO INHERIT in the parent, this is allowed.
*/
if (attribute->attnotnull && !childatt->attnotnull)
- ereport(ERROR,
- (errcode(ERRCODE_DATATYPE_MISMATCH),
- errmsg("column \"%s\" in child table must be marked NOT NULL",
- attributeName)));
+ {
+ HeapTuple contup;
+
+ contup = findNotNullConstraintAttnum(RelationGetRelid(parent_rel),
+ attribute->attnum);
+ if (!((Form_pg_constraint) GETSTRUCT(contup))->connoinherit)
+ ereport(ERROR,
+ (errcode(ERRCODE_DATATYPE_MISMATCH),
+ errmsg("column \"%s\" in child table must be marked NOT NULL",
+ attributeName)));
+ }
/*
* Child column must be generated if and only if parent column is.
@@ -15264,6 +15850,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);
@@ -15277,7 +15864,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);
@@ -15289,7 +15876,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 */
@@ -15309,22 +15897,49 @@ 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\"",
RelationGetRelationName(child_rel),
NameStr(parent_con->conname))));
- /* If the child constraint is "no inherit" then cannot merge */
- if (child_con->connoinherit)
+ /*
+ * If the CHECK child constraint is "no inherit" then cannot
+ * merge.
+ *
+ * This is not desirable for not-null constraints, mostly because
+ * it breaks our pg_upgrade strategy, but it also makes sense on
+ * its own: if a child has its own not-null constraint and then
+ * acquires a parent with the same constraint, then we start to
+ * enforce that constraint for all the descendants of that child
+ * too, if any.
+ */
+ if (child_con->contype == CONSTRAINT_CHECK &&
+ child_con->connoinherit)
ereport(ERROR,
(errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
errmsg("constraint \"%s\" conflicts with non-inherited constraint on child table \"%s\"",
@@ -15353,6 +15968,9 @@ MergeConstraintsIntoExisting(Relation child_rel, Relation parent_rel)
ereport(ERROR,
errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
errmsg("too many inheritance parents"));
+ if (child_con->contype == CONSTRAINT_NOTNULL &&
+ child_con->connoinherit)
+ child_con->connoinherit = false;
/*
* In case of partitions, an inherited constraint must be
@@ -15416,6 +16034,18 @@ ATExecDropInherit(Relation rel, RangeVar *parent, LOCKMODE lockmode)
/* Off to RemoveInheritance() where most of the work happens */
RemoveInheritance(rel, parent_rel, false);
+ /*
+ * If parent_rel has a primary key, then child_rel has not-null
+ * constraints that make these columns as non nullable. Mark those
+ * constraints as no longer inherited by this parent.
+ */
+ ATInheritAdjustNotNulls(parent_rel, rel, -1);
+
+ /*
+ * If the parent has a primary key, then we decrement counts for all NOT
+ * NULL constraints
+ */
+
ObjectAddressSet(address, RelationRelationId,
RelationGetRelid(parent_rel));
@@ -15524,6 +16154,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;
@@ -15594,6 +16225,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],
@@ -15604,6 +16237,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)))
{
@@ -15611,6 +16245,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);
@@ -15626,21 +16262,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)
{
@@ -15680,6 +16335,54 @@ RemoveInheritance(Relation child_rel, Relation parent_rel, bool expect_detached)
}
/*
+ * Adjust coninhcount of not-null constraints upwards or downwards when a
+ * table is marked as inheriting or no longer doing so a table with a primary
+ * key.
+ *
+ * Note: these constraints are not dropped, even if their inhcount goes to zero
+ * and conislocal is false. Instead we mark the constraints as locally defined.
+ * This is seen as more useful behavior, with no downsides. The user can always
+ * drop them afterwards.
+ */
+static void
+ATInheritAdjustNotNulls(Relation parent_rel, Relation child_rel, int inhcount)
+{
+ Bitmapset *pkattnos;
+
+ /* Quick exit when parent has no PK */
+ if (!parent_rel->rd_rel->relhasindex)
+ return;
+
+ pkattnos = RelationGetIndexAttrBitmap(parent_rel,
+ INDEX_ATTR_BITMAP_PRIMARY_KEY);
+ if (pkattnos != NULL)
+ {
+ Bitmapset *childattnums = NULL;
+ AttrMap *attmap;
+ int i;
+
+ attmap = build_attrmap_by_name(RelationGetDescr(parent_rel),
+ RelationGetDescr(child_rel), true);
+
+ i = -1;
+ while ((i = bms_next_member(pkattnos, i)) >= 0)
+ {
+ childattnums = bms_add_member(childattnums,
+ attmap->attnums[i + FirstLowInvalidHeapAttributeNumber - 1]);
+ }
+
+ /*
+ * CCI is needed in case there's a NOT NULL PRIMARY KEY column in the
+ * parent: the relevant not-null constraint in the child already had
+ * its inhcount modified earlier.
+ */
+ CommandCounterIncrement();
+ AdjustNotNullInheritance(RelationGetRelid(child_rel), childattnums,
+ inhcount);
+ }
+}
+
+/*
* Drop the dependency created by StoreCatalogInheritance1 (CREATE TABLE
* INHERITS/ALTER TABLE INHERIT -- refclassid will be RelationRelationId) or
* heap_create_with_catalog (CREATE TABLE OF/ALTER TABLE OF -- refclassid will
@@ -17530,7 +18233,7 @@ ComputePartitionAttrs(ParseState *pstate, Relation rel, List *partParams, AttrNu
* Do scanrel's existing constraints imply the partition constraint?
*
* "Existing constraints" include its check constraints and column-level
- * NOT NULL constraints. partConstraint describes the partition constraint,
+ * not-null constraints. partConstraint describes the partition constraint,
* in implicit-AND form.
*/
bool
@@ -17910,7 +18613,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);
@@ -18023,13 +18726,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;
@@ -18045,14 +18747,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++;
}
/*
@@ -18118,7 +18819,7 @@ AttachPartitionEnsureIndexes(Relation rel, Relation attachrel)
* the first matching, valid, 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;
@@ -18178,6 +18879,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,
@@ -18190,7 +18913,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);
@@ -18821,8 +19544,8 @@ DetachAddConstraintIfNeeded(List **wqueue, Relation partRel)
n->initially_valid = true;
n->skip_validation = true;
/* It's a re-add, since it nominally already exists */
- ATAddCheckConstraint(wqueue, tab, partRel, n,
- true, false, true, ShareUpdateExclusiveLock);
+ ATAddCheckNNConstraint(wqueue, tab, partRel, n,
+ true, false, true, ShareUpdateExclusiveLock);
}
}
@@ -19091,6 +19814,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))
@@ -19235,6 +19965,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 955286513d2..e66a99247e4 100644
--- a/src/backend/nodes/outfuncs.c
+++ b/src/backend/nodes/outfuncs.c
@@ -718,6 +718,11 @@ _outConstraint(StringInfo str, const Constraint *node)
case CONSTR_NOTNULL:
appendStringInfoString(str, "NOT_NULL");
+ WRITE_NODE_FIELD(keys);
+ WRITE_INT_FIELD(inhcount);
+ WRITE_BOOL_FIELD(is_no_inherit);
+ 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 97e43cbb49c..cc2021c1f7b 100644
--- a/src/backend/nodes/readfuncs.c
+++ b/src/backend/nodes/readfuncs.c
@@ -390,10 +390,17 @@ _readConstraint(void)
switch (local_node->contype)
{
case CONSTR_NULL:
- case CONSTR_NOTNULL:
/* no extra fields */
break;
+ case CONSTR_NOTNULL:
+ READ_NODE_FIELD(keys);
+ READ_INT_FIELD(inhcount);
+ READ_BOOL_FIELD(is_no_inherit);
+ 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 39932d3c2d2..243c8fb1e42 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 b3bdf947b6e..1b0e9e58195 100644
--- a/src/backend/parser/gram.y
+++ b/src/backend/parser/gram.y
@@ -3837,12 +3837,15 @@ ColConstraint:
* or be part of a_expr NOT LIKE or similar constructs).
*/
ColConstraintElem:
- NOT NULL_P
+ NOT NULL_P opt_no_inherit
{
Constraint *n = makeNode(Constraint);
n->contype = CONSTR_NOTNULL;
n->location = @1;
+ n->is_no_inherit = $3;
+ n->skip_validation = false;
+ n->initially_valid = true;
$$ = (Node *) n;
}
| NULL_P
@@ -4079,6 +4082,20 @@ ConstraintElem:
n->initially_valid = !n->skip_validation;
$$ = (Node *) n;
}
+ | NOT NULL_P ColId ConstraintAttributeSpec
+ {
+ Constraint *n = makeNode(Constraint);
+
+ n->contype = CONSTR_NOTNULL;
+ n->location = @1;
+ n->keys = list_make1(makeString($3));
+ /* no NOT VALID support yet */
+ processCASbits($4, @4, "NOT NULL",
+ NULL, NULL, NULL,
+ &n->is_no_inherit, yyscanner);
+ n->initially_valid = true;
+ $$ = (Node *) n;
+ }
| UNIQUE opt_unique_null_treatment '(' columnList ')' opt_c_include opt_definition OptConsTableSpace
ConstraintAttributeSpec
{
diff --git a/src/backend/parser/parse_utilcmd.c b/src/backend/parser/parse_utilcmd.c
index e48e9e99d36..bab7b87fe86 100644
--- a/src/backend/parser/parse_utilcmd.c
+++ b/src/backend/parser/parse_utilcmd.c
@@ -81,6 +81,7 @@ typedef struct
bool isalter; /* true if altering existing table */
List *columns; /* ColumnDef items */
List *ckconstraints; /* CHECK constraints */
+ List *nnconstraints; /* NOT NULL constraints */
List *fkconstraints; /* FOREIGN KEY constraints */
List *ixconstraints; /* index-creating constraints */
List *likeclauses; /* LIKE clauses that need post-processing */
@@ -240,6 +241,7 @@ transformCreateStmt(CreateStmt *stmt, const char *queryString)
cxt.isalter = false;
cxt.columns = NIL;
cxt.ckconstraints = NIL;
+ cxt.nnconstraints = NIL;
cxt.fkconstraints = NIL;
cxt.ixconstraints = NIL;
cxt.likeclauses = NIL;
@@ -346,6 +348,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);
@@ -535,6 +538,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);
@@ -632,10 +636,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... */
@@ -653,7 +655,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\"",
@@ -665,6 +667,10 @@ transformColumnDefinition(CreateStmtContext *cxt, ColumnDef *column)
break;
case CONSTR_NOTNULL:
+
+ /*
+ * Disallow conflicting [NOT] NULL markings
+ */
if (saw_nullable && !column->is_not_null)
ereport(ERROR,
(errcode(ERRCODE_SYNTAX_ERROR),
@@ -672,8 +678,25 @@ transformColumnDefinition(CreateStmtContext *cxt, ColumnDef *column)
column->colname, cxt->relation->relname),
parser_errposition(cxt->pstate,
constraint->location)));
- column->is_not_null = true;
- saw_nullable = true;
+ /* Ignore redundant NOT NULL markings */
+
+ /*
+ * If this is the first time we see this column being marked
+ * not null, add the constraint entry; and get rid of any
+ * previous markings to mark the column NOT NULL.
+ */
+ if (!column->is_not_null)
+ {
+ column->is_not_null = true;
+ saw_nullable = true;
+
+ constraint->keys = list_make1(makeString(column->colname));
+ cxt->nnconstraints = lappend(cxt->nnconstraints, constraint);
+
+ /* Don't need this anymore, if we had it */
+ need_notnull = false;
+ }
+
break;
case CONSTR_DEFAULT:
@@ -723,16 +746,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;
}
@@ -839,6 +865,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->keys = list_make1(makeString(column->colname));
+ notnull->skip_validation = false;
+ notnull->initially_valid = true;
+
+ cxt->nnconstraints = lappend(cxt->nnconstraints, notnull);
+ }
+
+ /*
* If needed, generate ALTER FOREIGN TABLE ALTER COLUMN statement to add
* per-column foreign data wrapper options to this column after creation.
*/
@@ -907,6 +956,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,
@@ -918,7 +971,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:
@@ -954,6 +1006,7 @@ transformTableLikeClause(CreateStmtContext *cxt, TableLikeClause *table_like_cla
AclResult aclresult;
char *comment;
ParseCallbackState pcbstate;
+ bool process_notnull_constraints = false;
setup_parser_errposition_callback(&pcbstate, cxt->pstate,
table_like_clause->relation->location);
@@ -1025,8 +1078,9 @@ transformTableLikeClause(CreateStmtContext *cxt, TableLikeClause *table_like_cla
/*
* Create a new column, which is marked as NOT inherited.
*
- * For constraints, ONLY the NOT NULL constraint is inherited by the
- * new column definition per SQL99.
+ * For constraints, ONLY the not-null constraint is inherited by the
+ * new column definition per SQL99; however we cannot do that
+ * correctly here, so we leave it for expandTableLikeClause to handle.
*/
def = makeNode(ColumnDef);
def->colname = pstrdup(attributeName);
@@ -1034,7 +1088,9 @@ transformTableLikeClause(CreateStmtContext *cxt, TableLikeClause *table_like_cla
attribute->atttypmod);
def->inhcount = 0;
def->is_local = true;
- def->is_not_null = attribute->attnotnull;
+ def->is_not_null = false;
+ if (attribute->attnotnull)
+ process_notnull_constraints = true;
def->is_from_type = false;
def->storage = 0;
def->raw_default = NULL;
@@ -1116,20 +1172,78 @@ 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 &
- (CREATE_TABLE_LIKE_DEFAULTS |
- CREATE_TABLE_LIKE_GENERATED |
- CREATE_TABLE_LIKE_CONSTRAINTS |
- CREATE_TABLE_LIKE_INDEXES))
+ if ((table_like_clause->options &
+ (CREATE_TABLE_LIKE_DEFAULTS |
+ CREATE_TABLE_LIKE_GENERATED |
+ CREATE_TABLE_LIKE_CONSTRAINTS |
+ CREATE_TABLE_LIKE_INDEXES)) ||
+ process_notnull_constraints)
{
table_like_clause->relationOid = RelationGetRelid(relation);
cxt->likeclauses = lappend(cxt->likeclauses, table_like_clause);
}
/*
+ * If INCLUDING INDEXES is not given and a primary key exists, we need to
+ * add not-null constraints to the columns covered by the PK (except those
+ * that already have one.) This is required for backwards compatibility.
+ */
+ if ((table_like_clause->options & CREATE_TABLE_LIKE_INDEXES) == 0)
+ {
+ Bitmapset *pkcols;
+ int x = -1;
+ Bitmapset *donecols = NULL;
+ ListCell *lc;
+
+ /*
+ * Obtain a bitmapset of columns on which we'll add not-null
+ * constraints in expandTableLikeClause, so that we skip this for
+ * those.
+ */
+ foreach(lc, RelationGetNotNullConstraints(RelationGetRelid(relation), true))
+ {
+ CookedConstraint *cooked = (CookedConstraint *) lfirst(lc);
+
+ donecols = bms_add_member(donecols, cooked->attnum);
+ }
+
+ pkcols = RelationGetIndexAttrBitmap(relation,
+ INDEX_ATTR_BITMAP_PRIMARY_KEY);
+ while ((x = bms_next_member(pkcols, x)) >= 0)
+ {
+ Constraint *notnull;
+ AttrNumber attnum = x + FirstLowInvalidHeapAttributeNumber;
+ Form_pg_attribute attForm;
+
+ /* ignore if we already have one for this column */
+ if (bms_is_member(attnum, donecols))
+ continue;
+
+ attForm = TupleDescAttr(tupleDesc, attnum - 1);
+
+ notnull = makeNode(Constraint);
+ notnull->contype = CONSTR_NOTNULL;
+ notnull->conname = NULL;
+ notnull->is_no_inherit = false;
+ notnull->deferrable = false;
+ notnull->initdeferred = false;
+ notnull->location = -1;
+ notnull->keys = list_make1(makeString(pstrdup(NameStr(attForm->attname))));
+ notnull->skip_validation = false;
+ notnull->initially_valid = true;
+
+ cxt->nnconstraints = lappend(cxt->nnconstraints, notnull);
+ }
+ }
+
+ /*
* We may copy extended statistics if requested, since the representation
* of CreateStatsStmt doesn't depend on column numbers.
*/
@@ -1195,6 +1309,8 @@ expandTableLikeClause(RangeVar *heapRel, TableLikeClause *table_like_clause)
TupleConstr *constr;
AttrMap *attmap;
char *comment;
+ bool at_pushed = false;
+ ListCell *lc;
/*
* Open the relation referenced by the LIKE clause. We should still have
@@ -1375,6 +1491,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(RelationGetRelid(relation), false))
+ {
+ AlterTableCmd *atsubcmd;
+
+ atsubcmd = makeNode(AlterTableCmd);
+ atsubcmd->subtype = AT_AddConstraint;
+ atsubcmd->def = (Node *) lfirst_node(Constraint, lc);
+ atsubcmds = lappend(atsubcmds, atsubcmd);
+ }
+
+ /*
* If we generated any ALTER TABLE actions above, wrap them into a single
* ALTER TABLE command. Stick it at the front of the result, so it runs
* before any CommentStmts we made above.
@@ -1388,6 +1518,8 @@ expandTableLikeClause(RangeVar *heapRel, TableLikeClause *table_like_clause)
atcmd->objtype = OBJECT_TABLE;
atcmd->missing_ok = false;
result = lcons(atcmd, result);
+
+ at_pushed = true;
}
/*
@@ -1415,6 +1547,39 @@ expandTableLikeClause(RangeVar *heapRel, TableLikeClause *table_like_clause)
attmap,
NULL);
+ /*
+ * The PK columns might not yet non-nullable, so make sure they
+ * become so.
+ */
+ if (index_stmt->primary)
+ {
+ foreach(lc, index_stmt->indexParams)
+ {
+ IndexElem *col = lfirst_node(IndexElem, lc);
+ AlterTableCmd *notnullcmd = makeNode(AlterTableCmd);
+
+ notnullcmd->subtype = AT_SetAttNotNull;
+ notnullcmd->name = pstrdup(col->name);
+ /* Luckily we can still add more AT-subcmds here */
+ atsubcmds = lappend(atsubcmds, notnullcmd);
+ }
+
+ /*
+ * If we had already put the AlterTableStmt into the output
+ * list, we don't need to do so again; otherwise do it.
+ */
+ if (!at_pushed)
+ {
+ AlterTableStmt *atcmd = makeNode(AlterTableStmt);
+
+ atcmd->relation = copyObject(heapRel);
+ atcmd->cmds = atsubcmds;
+ atcmd->objtype = OBJECT_TABLE;
+ atcmd->missing_ok = false;
+ result = lcons(atcmd, result);
+ }
+ }
+
/* Copy comment on index, if requested */
if (table_like_clause->options & CREATE_TABLE_LIKE_COMMENTS)
{
@@ -1505,8 +1670,8 @@ transformOfType(CreateStmtContext *cxt, TypeName *ofTypename)
* with the index there.
*
* Unlike transformIndexConstraint, we don't make any effort to force primary
- * key columns to be NOT NULL. The larger cloning process this is part of
- * should have cloned their NOT NULL status separately (and DefineIndex will
+ * key columns to be not-null. The larger cloning process this is part of
+ * should have cloned their not-null status separately (and DefineIndex will
* complain if that fails to happen).
*/
IndexStmt *
@@ -2051,10 +2216,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 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)
{
@@ -2128,9 +2295,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);
}
@@ -2138,12 +2303,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 not-null constraint.
*/
static IndexStmt *
transformIndexConstraint(Constraint *constraint, CreateStmtContext *cxt)
@@ -2401,7 +2564,7 @@ transformIndexConstraint(Constraint *constraint, CreateStmtContext *cxt)
* For UNIQUE and PRIMARY KEY, we just have a list of column names.
*
* Make sure referenced keys exist. If we are making a PRIMARY KEY index,
- * also make sure they are NOT NULL.
+ * also make sure they are not-null.
*/
else
{
@@ -2409,7 +2572,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;
@@ -2428,15 +2590,16 @@ 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
+ * 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)
@@ -2479,14 +2642,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;
}
}
@@ -2540,15 +2695,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);
}
@@ -3320,6 +3471,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;
@@ -3563,8 +3715,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))
{
@@ -3587,19 +3739,26 @@ transformAlterTableStmt(Oid relid, AlterTableStmt *stmt,
}
cxt.alist = NIL;
- /* Append any CHECK or FK constraints to the commands list */
+ /* Append any CHECK, NOT NULL or FK constraints to the commands list */
foreach(l, cxt.ckconstraints)
{
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);
}
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);
}
diff --git a/src/backend/utils/adt/ruleutils.c b/src/backend/utils/adt/ruleutils.c
index 03f2835c3f1..97b0ef22ac5 100644
--- a/src/backend/utils/adt/ruleutils.c
+++ b/src/backend/utils/adt/ruleutils.c
@@ -2490,6 +2490,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:
/*
diff --git a/src/backend/utils/cache/relcache.c b/src/backend/utils/cache/relcache.c
index 8e08ca1c680..7234cb3da69 100644
--- a/src/backend/utils/cache/relcache.c
+++ b/src/backend/utils/cache/relcache.c
@@ -4789,19 +4789,41 @@ RelationGetIndexList(Relation relation)
result = lappend_oid(result, index->indexrelid);
/*
- * Invalid, non-unique, non-immediate or predicate indexes aren't
- * interesting for either oid indexes or replication identity indexes,
- * so don't check them.
+ * Non-unique, non-immediate or predicate indexes aren't interesting
+ * for either oid indexes or replication identity indexes, so don't
+ * check them.
*/
- if (!index->indisvalid || !index->indisunique ||
+ if (!index->indisunique ||
!index->indimmediate ||
!heap_attisnull(htup, Anum_pg_index_indpred, NULL))
continue;
- /* remember primary key index if any */
- if (index->indisprimary)
+ /*
+ * Remember primary key index, if any. We do this only if the index
+ * is valid; but if the table is partitioned, then we do it even if
+ * it's invalid.
+ *
+ * The reason for returning invalid primary keys for foreign tables is
+ * because of pg_dump of NOT NULL constraints, and the fact that PKs
+ * remain marked invalid until the partitions' PKs are attached to it.
+ * If we make rd_pkindex invalid, then the attnotnull flag is reset
+ * after the PK is created, which causes the ALTER INDEX ATTACH
+ * PARTITION to fail with 'column ... is not marked NOT NULL'. With
+ * this, dropconstraint_internal() will believe that the columns must
+ * not have attnotnull reset, so the PKs-on-partitions can be attached
+ * correctly, until finally the PK-on-parent is marked valid.
+ *
+ * Also, this doesn't harm anything, because rd_pkindex is not a
+ * "real" index anyway, but a RELKIND_PARTITIONED_INDEX.
+ */
+ if (index->indisprimary &&
+ (index->indisvalid ||
+ relation->rd_rel->relkind == RELKIND_PARTITIONED_TABLE))
pkeyIndex = index->indexrelid;
+ if (!index->indisvalid)
+ continue;
+
/* remember explicitly chosen replica index */
if (index->indisreplident)
candidateIndex = index->indexrelid;
generated by cgit v1.2.3 (git 2.39.1) at 2025-07-31 17:41:13 +0000