database/postgres
1
0
Fork 0
mirror of https://github.com/postgres/postgres.git synced 2025-09-02 04:21:28 +03:00
Code Activity

Catalog not-null constraints

Browse Source 
We now create contype='n' pg_constraint rows for not-null constraints.

We propagate these constraints to other tables during operations such as
adding inheritance relationships, creating and attaching partitions and
creating tables LIKE other tables.  We also spawn not-null constraints
for inheritance child tables when their parents have primary keys.
These related constraints mostly follow the well-known rules of
conislocal and coninhcount that we have for CHECK constraints, with some
adaptations: for example, as opposed to CHECK constraints, we don't
match not-null ones by name when descending a hierarchy to alter it,
instead matching by column name that they apply to.  This means we don't
require the constraint names to be identical across a hierarchy.

For now, we omit them for system catalogs.  Maybe this is worth
reconsidering.  We don't support NOT VALID nor DEFERRABLE clauses
either; these can be added as separate features later (this patch is
already large and complicated enough.)

psql shows these constraints in \d+.

pg_dump requires some ad-hoc hacks, particularly when dumping a primary
key.  We now create one "throwaway" not-null constraint for each column
in the PK together with the CREATE TABLE command, and once the PK is
created, all those throwaway constraints are removed.  This avoids
having to check each tuple for nullness when the dump restores the
primary key creation.

pg_upgrading from an older release requires a somewhat brittle procedure
to create a constraint state that matches what would be created if the
database were being created fresh in Postgres 17.  I have tested all the
scenarios I could think of, and it works correctly as far as I can tell,
but I could have neglected weird cases.

This patch has been very long in the making.  The first patch was
written by Bernd Helmle in 2010 to add a new pg_constraint.contype value
('n'), which I (Álvaro) then hijacked in 2011 and 2012, until that one
was killed by the realization that we ought to use contype='c' instead:
manufactured CHECK constraints.  However, later SQL standard
development, as well as nonobvious emergent properties of that design
(mostly, failure to distinguish them from "normal" CHECK constraints as
well as the performance implication of having to test the CHECK
expression) led us to reconsider this choice, so now the current
implementation uses contype='n' again.  During Postgres 16 this had
already been introduced by commit e056c557ae, but there were some
problems mainly with the pg_upgrade procedure that couldn't be fixed in
reasonable time, so it was reverted.

In 2016 Vitaly Burovoy also worked on this feature[1] but found no
consensus for his proposed approach, which was claimed to be closer to
the letter of the standard, requiring an additional pg_attribute column
to track the OID of the not-null constraint for that column.
[1] https://postgr.es/m/CAKOSWNkN6HSyatuys8xZxzRCR-KL1OkHS5-b9qd9bf1Rad3PLA@mail.gmail.com

Author: Álvaro Herrera <alvherre@alvh.no-ip.org>
Author: Bernd Helmle <mailings@oopsware.de>
Reviewed-by: Justin Pryzby <pryzby@telsasoft.com>
Reviewed-by: Peter Eisentraut <peter.eisentraut@enterprisedb.com>
Reviewed-by: Dean Rasheed <dean.a.rasheed@gmail.com>
This commit is contained in:
Alvaro Herrera
2023-08-25 13:31:24 +02:00
parent 9c13b6814a
commit b0e96f3119
50 changed files with 3933 additions and 769 deletions
Download Patch File Download Diff File Expand all files Collapse all files

533
src/backend/catalog/heap.c
View File

@@ -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 */
@@ -2147,6 +2147,53 @@ StoreRelCheck(Relation rel, const char *ccname, Node *expr,
return constrOid;
}
/*
* 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).
*
@@ -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);
/*
* Here, we assume the parser will only pass us valid CHECK
* expressions, so we do no particular checking.
*/
expr = stringToNode(cdef->cooked_expr);
}
/*
* Check name uniqueness, or generate a name if none was given.
*/
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 (cdef->raw_expr != NULL)
{
if (strcmp((char *) lfirst(cell2), ccname) == 0)
ereport(ERROR,
(errcode(ERRCODE_DUPLICATE_OBJECT),
errmsg("check constraint \"%s\" already exists",
ccname)));
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);
}
/* 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.)
* Check name uniqueness, or generate a name if none was given.
*/
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;
if (cdef->conname != NULL)
{
ListCell *cell2;
vars = pull_var_clause(expr, 0);
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)));
}
/* eliminate duplicates */
vars = list_union(NIL, vars);
/* save name for future checks */
checknames = lappend(checknames, ccname);
if (list_length(vars) == 1)
colname = get_attname(RelationGetRelid(rel),
((Var *) linitial(vars))->varattno,
true);
/*
* 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
colname = NULL;
{
/*
* 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;
ccname = ChooseConstraintName(RelationGetRelationName(rel),
colname,
"check",
RelationGetNamespace(rel),
checknames);
vars = pull_var_clause(expr, 0);
/* save name for future checks */
checknames = lappend(checknames, ccname);
/* 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;
ccname = ChooseConstraintName(RelationGetRelationName(rel),
colname,
"check",
RelationGetNamespace(rel),
checknames);
/* save name for future checks */
checknames = lappend(checknames, ccname);
}
/*
* 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);
}
else if (cdef->contype == CONSTR_NOTNULL)
{
CookedConstraint *nncooked;
AttrNumber colnum;
char *nnname;
/*
* OK, store it.
*/
constrOid =
StoreRelCheck(rel, ccname, expr, cdef->initially_valid, is_local,
is_local ? 0 : 1, cdef->is_no_inherit, is_internal);
/* 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));
numchecks++;
/*
* 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;
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);
/*
* If a constraint name is specified, check that it isn't already
* used. Otherwise, choose a non-conflicting one ourselves.
*/
if (cdef->conname)
{
if (ConstraintNameIsUsed(CONSTRAINT_RELATION,
RelationGetRelid(rel),
cdef->conname))
ereport(ERROR,
errcode(ERRCODE_DUPLICATE_OBJECT),
errmsg("constraint \"%s\" for relation \"%s\" already exists",
cdef->conname, RelationGetRelationName(rel)));
nnname = cdef->conname;
}
else
nnname = ChooseConstraintName(RelationGetRelationName(rel),
strVal(linitial(cdef->keys)),
"not_null",
RelationGetNamespace(rel),
nnnames);
nnnames = lappend(nnnames, nnname);
constrOid =
StoreRelNotNull(rel, nnname, colnum,
cdef->initially_valid,
cdef->inhcount == 0,
cdef->inhcount,
cdef->is_no_inherit);
nncooked = (CookedConstraint *) palloc(sizeof(CookedConstraint));
nncooked->contype = CONSTR_NOTNULL;
nncooked->conoid = constrOid;
nncooked->name = nnname;
nncooked->attnum = colnum;
nncooked->expr = NULL;
nncooked->skip_validation = cdef->skip_validation;
nncooked->is_local = is_local;
nncooked->inhcount = cdef->inhcount;
nncooked->is_no_inherit = cdef->is_no_inherit;
cookedConstraints = lappend(cookedConstraints, nncooked);
}
}
/*
@@ -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.
*

286
src/backend/catalog/pg_constraint.c
View File

@@ -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"
@@ -562,6 +563,291 @@ ChooseConstraintName(const char *name1, const char *name2,
return conname;
}
/*
* 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.
*/

1601
src/backend/commands/tablecmds.c
View File

File diff suppressed because it is too large Load Diff

5
src/backend/nodes/outfuncs.c
View File

@@ -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:

9
src/backend/nodes/readfuncs.c
View File

@@ -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);

2
src/backend/optimizer/util/plancat.c
View File

@@ -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);

19
src/backend/parser/gram.y
View File

@@ -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
{

277
src/backend/parser/parse_utilcmd.c
View File

@@ -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;
}
@@ -838,6 +864,29 @@ transformColumnDefinition(CreateStmtContext *cxt, ColumnDef *column)
constraint->location)));
}
/*
* 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,19 +1172,77 @@ 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
@@ -1374,6 +1490,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
@@ -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);
}

14
src/backend/utils/adt/ruleutils.c
View File

@@ -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:
/*

34
src/backend/utils/cache/relcache.c vendored
View File

@@ -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;