database/postgres
1
0
Fork 0
mirror of https://github.com/postgres/postgres.git synced 2025-11-22 12:22:45 +03:00
Code Activity

Add pg_constraint rows for not-null constraints

Browse Source 
We now create contype='n' pg_constraint rows for not-null constraints on
user tables.  Only one such constraint is allowed for a column.

We propagate these constraints to other tables during operations such as
adding inheritance relationships, creating and attaching partitions and
creating tables LIKE other tables.  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 or remove it, instead matching by the name of the
column that they apply to.  This means we don't require the constraint
names to be identical across a hierarchy.

The inheritance status of these constraints can be controlled: now we
can be sure that if a parent table has one, then all children will have
it as well.  They can optionally be marked NO INHERIT, and then children
are free not to have one.  (There's currently no support for altering a
NO INHERIT constraint into inheriting down the hierarchy, but that's a
desirable future feature.)

This also opens the door for having these constraints be marked NOT
VALID, as well as allowing UNIQUE+NOT NULL to be used for functional
dependency determination, as envisioned by commit e49ae8d3bc.  It's
likely possible to allow DEFERRABLE constraints as followup work, as
well.

psql shows these constraints in \d+, though we may want to reconsider if
this turns out to be too noisy.  Earlier versions of this patch hid
constraints that were on the same columns of the primary key, but I'm
not sure that that's very useful.  If clutter is a problem, we might be
better off inventing a new \d++ command and not showing the constraints
in \d+.

For now, we omit these constraints on system catalog columns, because
they're unlikely to achieve anything.

The main difference to the previous attempt at this (b0e96f3119) is
that we now require that such a constraint always exists when a primary
key is in the column; we didn't require this previously which had a
number of unpalatable consequences.  With this requirement, the code is
easier to reason about.  For example:

- We no longer have "throwaway constraints" during pg_dump.  We needed
  those for the case where a table had a PK without a not-null
  underneath, to prevent a slow scan of the data during restore of the
  PK creation, which was particularly problematic for pg_upgrade.

- We no longer have to cope with attnotnull being set spuriously in
  case a primary key is dropped indirectly (e.g., via DROP COLUMN).

Some bits of code in this patch were authored by Jian He.

Author: Álvaro Herrera <alvherre@alvh.no-ip.org>
Author: Bernd Helmle <mailings@oopsware.de>
Reviewed-by: 何建 (jian he) <jian.universality@gmail.com>
Reviewed-by: 王刚 (Tender Wang) <tndrwang@gmail.com>
Reviewed-by: Justin Pryzby <pryzby@telsasoft.com>
Reviewed-by: Peter Eisentraut <peter.eisentraut@enterprisedb.com>
Reviewed-by: Dean Rasheed <dean.a.rasheed@gmail.com>
Discussion: https://postgr.es/m/202408310358.sdhumtyuy2ht@alvherre.pgsql
This commit is contained in:
Álvaro Herrera
2024-11-08 13:28:48 +01:00
parent 075acdd933
commit 14e87ffa5c
57 changed files with 4257 additions and 829 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -2173,6 +2173,56 @@ 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;
Assert(attnum > InvalidAttrNumber);
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,
false);
return constrOid;
}
/*
* Store defaults and constraints (passed as a list of CookedConstraint).
*
@@ -2217,6 +2267,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);
@@ -2245,7 +2303,7 @@ StoreConstraints(Relation rel, List *cooked_constraints, bool is_internal)
* cooked CHECK constraints
*
* All entries in newColDefaults will be processed. Entries in newConstraints
* will be processed only if they are CONSTR_CHECK type.
* will be processed only if they are CONSTR_CHECK or CONSTR_NOTNULL types.
*
* Returns a list of CookedConstraint nodes that shows the cooked form of
* the default and constraint expressions added to the relation.
@@ -2274,6 +2332,7 @@ AddRelationNewConstraints(Relation rel,
ParseNamespaceItem *nsitem;
int numchecks;
List *checknames;
List *nnnames;
Node *expr;
CookedConstraint *cooked;
@@ -2358,6 +2417,7 @@ AddRelationNewConstraints(Relation rel,
*/
numchecks = numoldchecks;
checknames = NIL;
nnnames = NIL;
foreach_node(Constraint, cdef, newConstraints)
{
Oid constrOid;
@@ -2412,7 +2472,7 @@ AddRelationNewConstraints(Relation rel,
* Check against pre-existing constraints. If we are allowed
* to merge with an existing constraint, there's no more to do
* here. (We omit the duplicate constraint from the result,
* which is what ATAddCheckConstraint wants.)
* which is what ATAddCheckNNConstraint wants.)
*/
if (MergeWithExistingConstraint(rel, ccname, expr,
allow_merge, is_local,
@@ -2481,6 +2541,77 @@ AddRelationNewConstraints(Relation rel,
cooked->is_no_inherit = cdef->is_no_inherit;
cookedConstraints = lappend(cookedConstraints, cooked);
}
else if (cdef->contype == CONSTR_NOTNULL)
{
CookedConstraint *nncooked;
AttrNumber colnum;
int16 inhcount = is_local ? 0 : 1;
char *nnname;
/* Determine which column to modify */
colnum = get_attnum(RelationGetRelid(rel), strVal(linitial(cdef->keys)));
if (colnum == InvalidAttrNumber)
ereport(ERROR,
errcode(ERRCODE_UNDEFINED_COLUMN),
errmsg("column \"%s\" of relation \"%s\" does not exist",
strVal(linitial(cdef->keys)), RelationGetRelationName(rel)));
if (colnum < InvalidAttrNumber)
ereport(ERROR,
errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
errmsg("cannot add not-null constraint on system column \"%s\"",
strVal(linitial(cdef->keys))));
/*
* If the column already has a not-null constraint, we don't want
* to add another one; just adjust inheritance status as needed.
*/
if (AdjustNotNullInheritance(RelationGetRelid(rel), colnum,
is_local, cdef->is_no_inherit))
continue;
/*
* If a constraint name is specified, check that it isn't already
* used. Otherwise, choose a non-conflicting one ourselves.
*/
if (cdef->conname)
{
if (ConstraintNameIsUsed(CONSTRAINT_RELATION,
RelationGetRelid(rel),
cdef->conname))
ereport(ERROR,
errcode(ERRCODE_DUPLICATE_OBJECT),
errmsg("constraint \"%s\" for relation \"%s\" already exists",
cdef->conname, RelationGetRelationName(rel)));
nnname = cdef->conname;
}
else
nnname = ChooseConstraintName(RelationGetRelationName(rel),
strVal(linitial(cdef->keys)),
"not_null",
RelationGetNamespace(rel),
nnnames);
nnnames = lappend(nnnames, nnname);
constrOid =
StoreRelNotNull(rel, nnname, colnum,
cdef->initially_valid,
is_local,
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 = inhcount;
nncooked->is_no_inherit = cdef->is_no_inherit;
cookedConstraints = lappend(cookedConstraints, nncooked);
}
}
/*
@@ -2648,6 +2779,262 @@ MergeWithExistingConstraint(Relation rel, const char *ccname, Node *expr,
return found;
}
/*
* 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.
*
* 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;
/*
* 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.
*
* 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.
*/
for (int outerpos = 0; outerpos < list_length(constraints); outerpos++)
{
Constraint *constr;
AttrNumber attnum;
char *conname;
int inhcount = 0;
constr = list_nth_node(Constraint, constraints, outerpos);
Assert(constr->contype == CONSTR_NOTNULL);
attnum = get_attnum(RelationGetRelid(rel),
strVal(linitial(constr->keys)));
if (attnum == InvalidAttrNumber)
ereport(ERROR,
errcode(ERRCODE_UNDEFINED_COLUMN),
errmsg("column \"%s\" of relation \"%s\" does not exist",
strVal(linitial(constr->keys)),
RelationGetRelationName(rel)));
if (attnum < InvalidAttrNumber)
ereport(ERROR,
errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
errmsg("cannot add not-null constraint on system column \"%s\"",
strVal(linitial(constr->keys))));
/*
* A column can only have one not-null constraint, so discard any
* additional ones that appear for columns we already saw; but check
* that the NO INHERIT flags match.
*/
for (int restpos = outerpos + 1; restpos < list_length(constraints);)
{
Constraint *other;
other = list_nth_node(Constraint, constraints, restpos);
if (strcmp(strVal(linitial(constr->keys)),
strVal(linitial(other->keys))) == 0)
{
if (other->is_no_inherit != constr->is_no_inherit)
ereport(ERROR,
errcode(ERRCODE_SYNTAX_ERROR),
errmsg("conflicting NO INHERIT declaration for not-null constraint on column \"%s\"",
strVal(linitial(constr->keys))));
/*
* Preserve constraint name if one is specified, but raise an
* error if conflicting ones are specified.
*/
if (other->conname)
{
if (!constr->conname)
constr->conname = pstrdup(other->conname);
else if (strcmp(constr->conname, other->conname) != 0)
ereport(ERROR,
errcode(ERRCODE_SYNTAX_ERROR),
errmsg("conflicting not-null constraint names \"%s\" and \"%s\"",
constr->conname, other->conname));
}
/* XXX do we need to verify any other fields? */
constraints = list_delete_nth_cell(constraints, restpos);
}
else
restpos++;
}
/*
* Search in the list of inherited constraints for any entries on the
* same column; determine an inheritance count from that. Also, if at
* least one parent has a constraint for this column, then we must not
* accept a user specification for a NO INHERIT one. Any constraint
* from parents that we process here is deleted from the list: we no
* longer need to process it in the loop below.
*/
foreach_ptr(CookedConstraint, old, old_notnulls)
{
if (old->attnum == attnum)
{
/*
* If we get a constraint from the parent, having a local NO
* INHERIT one doesn't work.
*/
if (constr->is_no_inherit)
ereport(ERROR,
(errcode(ERRCODE_DATATYPE_MISMATCH),
errmsg("cannot define not-null constraint on column \"%s\" with NO INHERIT",
strVal(linitial(constr->keys))),
errdetail("The column has an inherited not-null constraint.")));
inhcount++;
old_notnulls = foreach_delete_current(old_notnulls, old);
}
}
/*
* 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_ptr(char, thisname, givennames)
{
if (strcmp(thisname, 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, true,
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 for that column. Because multiple
* parents could specify a not-null constraint for the same column, we
* must count how many there are and set an appropriate inhcount
* accordingly, deleting elements we've already processed.
*
* 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.
*/
for (int outerpos = 0; outerpos < list_length(old_notnulls); outerpos++)
{
CookedConstraint *cooked;
char *conname = NULL;
int inhcount = 1;
cooked = (CookedConstraint *) list_nth(old_notnulls, outerpos);
Assert(cooked->contype == CONSTR_NOTNULL);
Assert(cooked->name);
/*
* Preserve the first non-conflicting constraint name we come across.
*/
if (conname == NULL)
conname = cooked->name;
for (int restpos = outerpos + 1; restpos < list_length(old_notnulls);)
{
CookedConstraint *other;
other = (CookedConstraint *) list_nth(old_notnulls, restpos);
Assert(other->name);
if (other->attnum == cooked->attnum)
{
if (conname == NULL)
conname = other->name;
inhcount++;
old_notnulls = list_delete_nth_cell(old_notnulls, restpos);
}
else
restpos++;
}
/* If we got a name, make sure it isn't one we've already used */
if (conname != NULL)
{
foreach_ptr(char, thisname, nnnames)
{
if (strcmp(thisname, 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);
/* ignore the origin constraint's is_local and inhcount */
StoreRelNotNull(rel, conname, cooked->attnum, true,
false, inhcount, false);
nncols = lappend_int(nncols, cooked->attnum);
}
return nncols;
}
/*
* Update the count of constraints in the relation's pg_class tuple.
*

71
src/backend/catalog/information_schema.sql
View File

@@ -440,9 +440,8 @@ CREATE VIEW check_constraints AS
WHERE pg_has_role(coalesce(c.relowner, t.typowner), 'USAGE')
AND con.contype = 'c'
UNION
-- not-null constraints on domains
UNION ALL
-- not-null constraints
SELECT current_database()::information_schema.sql_identifier AS constraint_catalog,
rs.nspname::information_schema.sql_identifier AS constraint_schema,
con.conname::information_schema.sql_identifier AS constraint_name,
@@ -453,24 +452,7 @@ CREATE VIEW check_constraints AS
LEFT JOIN pg_type t ON t.oid = con.contypid
LEFT JOIN pg_attribute at ON (con.conrelid = at.attrelid AND con.conkey[1] = at.attnum)
WHERE pg_has_role(coalesce(c.relowner, t.typowner), 'USAGE'::text)
AND con.contype = 'n'
UNION
-- not-null constraints on relations
SELECT CAST(current_database() AS sql_identifier) AS constraint_catalog,
CAST(n.nspname AS sql_identifier) AS constraint_schema,
CAST(CAST(n.oid AS text) || '_' || CAST(r.oid AS text) || '_' || CAST(a.attnum AS text) || '_not_null' AS sql_identifier) AS constraint_name, -- XXX
CAST(a.attname || ' IS NOT NULL' AS character_data)
AS check_clause
FROM pg_namespace n, pg_class r, pg_attribute a
WHERE n.oid = r.relnamespace
AND r.oid = a.attrelid
AND a.attnum > 0
AND NOT a.attisdropped
AND a.attnotnull
AND r.relkind IN ('r', 'p')
AND pg_has_role(r.relowner, 'USAGE');
AND con.contype = 'n';
GRANT SELECT ON check_constraints TO PUBLIC;
@@ -839,6 +821,20 @@ CREATE VIEW constraint_column_usage AS
UNION ALL
/* not-null constraints */
SELECT DISTINCT nr.nspname, r.relname, r.relowner, a.attname, nc.nspname, c.conname
FROM pg_namespace nr, pg_class r, pg_attribute a, pg_namespace nc, pg_constraint c
WHERE nr.oid = r.relnamespace
AND r.oid = a.attrelid
AND r.oid = c.conrelid
AND a.attnum = c.conkey[1]
AND c.connamespace = nc.oid
AND c.contype = 'n'
AND r.relkind in ('r', 'p')
AND not a.attisdropped
UNION ALL
/* unique/primary key/foreign key constraints */
SELECT nr.nspname, r.relname, r.relowner, a.attname, nc.nspname, c.conname
FROM pg_namespace nr, pg_class r, pg_attribute a, pg_namespace nc,
@@ -1839,6 +1835,7 @@ CREATE VIEW table_constraints AS
CAST(r.relname AS sql_identifier) AS table_name,
CAST(
CASE c.contype WHEN 'c' THEN 'CHECK'
WHEN 'n' THEN 'CHECK'
WHEN 'f' THEN 'FOREIGN KEY'
WHEN 'p' THEN 'PRIMARY KEY'
WHEN 'u' THEN 'UNIQUE' END
@@ -1863,38 +1860,6 @@ CREATE VIEW table_constraints AS
AND c.contype NOT IN ('t', 'x') -- ignore nonstandard constraints
AND r.relkind IN ('r', 'p')
AND (NOT pg_is_other_temp_schema(nr.oid))
AND (pg_has_role(r.relowner, 'USAGE')
-- SELECT privilege omitted, per SQL standard
OR has_table_privilege(r.oid, 'INSERT, UPDATE, DELETE, TRUNCATE, REFERENCES, TRIGGER')
OR has_any_column_privilege(r.oid, 'INSERT, UPDATE, REFERENCES') )
UNION ALL
-- not-null constraints
SELECT CAST(current_database() AS sql_identifier) AS constraint_catalog,
CAST(nr.nspname AS sql_identifier) AS constraint_schema,
CAST(CAST(nr.oid AS text) || '_' || CAST(r.oid AS text) || '_' || CAST(a.attnum AS text) || '_not_null' AS sql_identifier) AS constraint_name, -- XXX
CAST(current_database() AS sql_identifier) AS table_catalog,
CAST(nr.nspname AS sql_identifier) AS table_schema,
CAST(r.relname AS sql_identifier) AS table_name,
CAST('CHECK' AS character_data) AS constraint_type,
CAST('NO' AS yes_or_no) AS is_deferrable,
CAST('NO' AS yes_or_no) AS initially_deferred,
CAST('YES' AS yes_or_no) AS enforced,
CAST(NULL AS yes_or_no) AS nulls_distinct
FROM pg_namespace nr,
pg_class r,
pg_attribute a
WHERE nr.oid = r.relnamespace
AND r.oid = a.attrelid
AND a.attnotnull
AND a.attnum > 0
AND NOT a.attisdropped
AND r.relkind IN ('r', 'p')
AND (NOT pg_is_other_temp_schema(nr.oid))
AND (pg_has_role(r.relowner, 'USAGE')
-- SELECT privilege omitted, per SQL standard
OR has_table_privilege(r.oid, 'INSERT, UPDATE, DELETE, TRUNCATE, REFERENCES, TRIGGER')

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

@@ -21,6 +21,7 @@
#include "access/table.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"
@@ -564,6 +565,78 @@ ChooseConstraintName(const char *name1, const char *name2,
return conname;
}
/*
* Find and return a copy of the pg_constraint tuple that implements a
* validated not-null constraint for the given column of the given relation.
* If no such constraint exists, return NULL.
*
* 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. If
* no such column or no such constraint exists, return NULL.
*/
HeapTuple
findNotNullConstraint(Oid relid, const char *colname)
{
AttrNumber attnum;
attnum = get_attnum(relid, colname);
if (attnum <= InvalidAttrNumber)
return NULL;
return findNotNullConstraintAttnum(relid, attnum);
}
/*
* Find and return the pg_constraint tuple that implements a validated
* not-null constraint for the given domain.
@@ -608,6 +681,185 @@ findDomainNotNullConstraint(Oid typid)
return retval;
}
/*
* 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));
Assert(colnum > 0 && colnum <= MaxAttrNumber);
if ((Pointer) arr != DatumGetPointer(adatum))
pfree(arr); /* free de-toasted copy, if any */
return colnum;
}
/*
* AdjustNotNullInheritance
* Adjust inheritance status for a single not-null constraint
*
* If no not-null constraint is found for the column, return false.
* Caller can create one.
* If a constraint exists but the connoinherit flag is not what the caller
* wants, throw an error about the incompatibility. Otherwise, we adjust
* conislocal/coninhcount and return true.
* In the latter case, if is_local is true we flip conislocal true, or do
* nothing if it's already true; otherwise we increment coninhcount by 1.
*/
bool
AdjustNotNullInheritance(Oid relid, AttrNumber attnum,
bool is_local, bool is_no_inherit)
{
HeapTuple tup;
tup = findNotNullConstraintAttnum(relid, attnum);
if (HeapTupleIsValid(tup))
{
Relation pg_constraint;
Form_pg_constraint conform;
bool changed = false;
pg_constraint = table_open(ConstraintRelationId, RowExclusiveLock);
conform = (Form_pg_constraint) GETSTRUCT(tup);
/*
* If the NO INHERIT flag we're asked for doesn't match what the
* existing constraint has, throw an error.
*/
if (is_no_inherit != conform->connoinherit)
ereport(ERROR,
errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
errmsg("cannot change NO INHERIT status of NOT NULL constraint \"%s\" on relation \"%s\"",
NameStr(conform->conname), get_rel_name(relid)));
if (!is_local)
{
if (pg_add_s16_overflow(conform->coninhcount, 1,
&conform->coninhcount))
ereport(ERROR,
errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
errmsg("too many inheritance parents"));
changed = true;
}
else if (!conform->conislocal)
{
conform->conislocal = true;
changed = true;
}
if (changed)
CatalogTupleUpdate(pg_constraint, &tup->t_self, tup);
table_close(pg_constraint, RowExclusiveLock);
return true;
}
return false;
}
/*
* 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.
*
* 'include_noinh' determines whether to include NO INHERIT constraints or not.
*/
List *
RelationGetNotNullConstraints(Oid relid, bool cooked, bool include_noinh)
{
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 && !include_noinh)
continue;
colnum = extractNotNullColumn(htup);
if (cooked)
{
CookedConstraint *cooked;
cooked = (CookedConstraint *) palloc(sizeof(CookedConstraint));
cooked->contype = CONSTR_NOTNULL;
cooked->conoid = conForm->oid;
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;
constr->is_no_inherit = conForm->connoinherit;
notnulls = lappend(notnulls, constr);
}
}
systable_endscan(conscan);
table_close(constrRel, AccessShareLock);
return notnulls;
}
/*
* Delete a single constraint record.
*/

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

File diff suppressed because it is too large Load Diff

4
src/backend/commands/typecmds.c
View File

@@ -944,6 +944,10 @@ DefineDomain(CreateDomainStmt *stmt)
ereport(ERROR,
(errcode(ERRCODE_SYNTAX_ERROR),
errmsg("conflicting NULL/NOT NULL constraints")));
if (constr->is_no_inherit)
ereport(ERROR,
errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
errmsg("not-null constraints for domains cannot be marked NO INHERIT"));
typNotNull = true;
nullDefined = true;
break;

23
src/backend/nodes/makefuncs.c
View File

@@ -436,6 +436,29 @@ makeRangeVar(char *schemaname, char *relname, int location)
return r;
}
/*
* makeNotNullConstraint -
* creates a Constraint node for NOT NULL constraints
*/
Constraint *
makeNotNullConstraint(String *colname)
{
Constraint *notnull;
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(colname);
notnull->skip_validation = false;
notnull->initially_valid = true;
return notnull;
}
/*
* makeTypeName -
* build a TypeName node for an unqualified name.

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

@@ -1698,6 +1698,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);

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

@@ -3908,12 +3908,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
@@ -4150,6 +4153,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_without_overlaps ')' opt_c_include opt_definition OptConsTableSpace
ConstraintAttributeSpec
{
@@ -4317,10 +4334,10 @@ DomainConstraintElem:
n->contype = CONSTR_NOTNULL;
n->location = @1;
n->keys = list_make1(makeString("value"));
/* no NOT VALID support yet */
/* no NOT VALID, NO INHERIT support */
processCASbits($3, @3, "NOT NULL",
NULL, NULL, NULL,
&n->is_no_inherit, yyscanner);
NULL, yyscanner);
n->initially_valid = true;
$$ = (Node *) n;
}

342
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;
@@ -303,6 +305,32 @@ transformCreateStmt(CreateStmt *stmt, const char *queryString)
Assert(stmt->constraints == NIL);
/*
* Before processing index constraints, which could include a primary key,
* we must scan all not-null constraints to propagate the is_not_null flag
* to each corresponding ColumnDef. This is necessary because table-level
* not-null constraints have not been marked in each ColumnDef, and the PK
* processing code needs to know whether one constraint has already been
* declared in order not to declare a redundant one.
*/
foreach_node(Constraint, nn, cxt.nnconstraints)
{
char *colname = strVal(linitial(nn->keys));
foreach_node(ColumnDef, cd, cxt.columns)
{
/* not our column? */
if (strcmp(cd->colname, colname) != 0)
continue;
/* Already marked not-null? Nothing to do */
if (cd->is_not_null)
break;
/* Bingo, we're done for this constraint */
cd->is_not_null = true;
break;
}
}
/*
* Postprocess constraints that give rise to index definitions.
*/
@@ -340,6 +368,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);
@@ -566,7 +595,9 @@ transformColumnDefinition(CreateStmtContext *cxt, ColumnDef *column)
bool saw_default;
bool saw_identity;
bool saw_generated;
ListCell *clist;
bool need_notnull = false;
bool disallow_noinherit_notnull = false;
Constraint *notnull_constraint = NULL;
cxt->columns = lappend(cxt->columns, column);
@@ -663,28 +694,54 @@ 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;
disallow_noinherit_notnull = true;
}
/* Process column constraints, if any... */
transformConstraintAttrs(cxt, column->constraints);
/*
* First, scan the column's constraints to see if a not-null constraint
* that we add must be prevented from being NO INHERIT. This should be
* enforced only for PRIMARY KEY, not IDENTITY or SERIAL. However, if the
* not-null constraint is specified as a table constraint rather than as a
* column constraint, AddRelationNotNullConstraints would raise an error
* if a NO INHERIT mismatch is found. To avoid inconsistently disallowing
* it in the table constraint case but not the column constraint case, we
* disallow it here as well. Maybe AddRelationNotNullConstraints can be
* improved someday, so that it doesn't complain, and then we can remove
* the restriction for SERIAL and IDENTITY here as well.
*/
if (!disallow_noinherit_notnull)
{
foreach_node(Constraint, constraint, column->constraints)
{
switch (constraint->contype)
{
case CONSTR_IDENTITY:
case CONSTR_PRIMARY:
disallow_noinherit_notnull = true;
break;
default:
break;
}
}
}
/* Now scan them again to do full processing */
saw_nullable = false;
saw_default = false;
saw_identity = false;
saw_generated = false;
foreach(clist, column->constraints)
foreach_node(Constraint, constraint, column->constraints)
{
Constraint *constraint = lfirst_node(Constraint, clist);
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\"",
@@ -696,6 +753,12 @@ transformColumnDefinition(CreateStmtContext *cxt, ColumnDef *column)
break;
case CONSTR_NOTNULL:
if (cxt->ispartitioned && constraint->is_no_inherit)
ereport(ERROR,
errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
errmsg("not-null constraints on partitioned tables cannot be NO INHERIT"));
/* Disallow conflicting [NOT] NULL markings */
if (saw_nullable && !column->is_not_null)
ereport(ERROR,
(errcode(ERRCODE_SYNTAX_ERROR),
@@ -703,8 +766,52 @@ transformColumnDefinition(CreateStmtContext *cxt, ColumnDef *column)
column->colname, cxt->relation->relname),
parser_errposition(cxt->pstate,
constraint->location)));
column->is_not_null = true;
saw_nullable = true;
if (disallow_noinherit_notnull && constraint->is_no_inherit)
ereport(ERROR,
errcode(ERRCODE_SYNTAX_ERROR),
errmsg("conflicting NO INHERIT declarations for not-null constraints on column \"%s\"",
column->colname));
/*
* If this is the first time we see this column being marked
* not-null, add the constraint entry and keep track of it.
* Also, remove previous markings that we need one.
*
* If this is a redundant not-null specification, just check
* that it doesn't conflict with what was specified earlier.
*
* Any conflicts with table constraints will be further
* checked in AddRelationNotNullConstraints().
*/
if (!column->is_not_null)
{
column->is_not_null = true;
saw_nullable = true;
need_notnull = false;
constraint->keys = list_make1(makeString(column->colname));
notnull_constraint = constraint;
cxt->nnconstraints = lappend(cxt->nnconstraints, constraint);
}
else if (notnull_constraint)
{
if (constraint->conname &&
notnull_constraint->conname &&
strcmp(notnull_constraint->conname, constraint->conname) != 0)
elog(ERROR, "conflicting not-null constraint names \"%s\" and \"%s\"",
notnull_constraint->conname, constraint->conname);
if (notnull_constraint->is_no_inherit != constraint->is_no_inherit)
ereport(ERROR,
errcode(ERRCODE_SYNTAX_ERROR),
errmsg("conflicting NO INHERIT declarations for not-null constraints on column \"%s\"",
column->colname));
if (!notnull_constraint->conname && constraint->conname)
notnull_constraint->conname = constraint->conname;
}
break;
case CONSTR_DEFAULT:
@@ -754,16 +861,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;
}
@@ -790,6 +900,15 @@ transformColumnDefinition(CreateStmtContext *cxt, ColumnDef *column)
break;
case CONSTR_PRIMARY:
if (saw_nullable && !column->is_not_null)
ereport(ERROR,
(errcode(ERRCODE_SYNTAX_ERROR),
errmsg("conflicting NULL/NOT NULL declarations for column \"%s\" of table \"%s\"",
column->colname, cxt->relation->relname),
parser_errposition(cxt->pstate,
constraint->location)));
need_notnull = true;
if (cxt->isforeign)
ereport(ERROR,
(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
@@ -869,6 +988,17 @@ transformColumnDefinition(CreateStmtContext *cxt, ColumnDef *column)
constraint->location)));
}
/*
* If we need a not-null constraint for PRIMARY KEY, SERIAL or IDENTITY,
* and one was not explicitly specified, add one now.
*/
if (need_notnull && !(saw_nullable && column->is_not_null))
{
column->is_not_null = true;
notnull_constraint = makeNotNullConstraint(makeString(column->colname));
cxt->nnconstraints = lappend(cxt->nnconstraints, notnull_constraint);
}
/*
* If needed, generate ALTER FOREIGN TABLE ALTER COLUMN statement to add
* per-column foreign data wrapper options to this column after creation.
@@ -938,6 +1068,15 @@ transformTableConstraint(CreateStmtContext *cxt, Constraint *constraint)
cxt->ckconstraints = lappend(cxt->ckconstraints, constraint);
break;
case CONSTR_NOTNULL:
if (cxt->ispartitioned && constraint->is_no_inherit)
ereport(ERROR,
errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
errmsg("not-null constraints on partitioned tables cannot be NO INHERIT"));
cxt->nnconstraints = lappend(cxt->nnconstraints, constraint);
break;
case CONSTR_FOREIGN:
if (cxt->isforeign)
ereport(ERROR,
@@ -949,7 +1088,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:
@@ -1053,14 +1191,10 @@ transformTableLikeClause(CreateStmtContext *cxt, TableLikeClause *table_like_cla
continue;
/*
* 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.
* Create a new column definition
*/
def = makeColumnDef(NameStr(attribute->attname), attribute->atttypid,
attribute->atttypmod, attribute->attcollation);
def->is_not_null = attribute->attnotnull;
/*
* Add to column list
@@ -1129,14 +1263,28 @@ transformTableLikeClause(CreateStmtContext *cxt, TableLikeClause *table_like_cla
}
}
/*
* Reproduce not-null constraints, if any, by copying them. We do this
* regardless of options given.
*/
if (tupleDesc->constr && tupleDesc->constr->has_not_null)
{
List *lst;
lst = RelationGetNotNullConstraints(RelationGetRelid(relation), false,
true);
cxt->nnconstraints = list_concat(cxt->nnconstraints, lst);
}
/*
* We cannot yet deal with defaults, CHECK constraints, indexes, or
* statistics, since 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 expandTableLikeClause is certain to
* open the same table.
* expandTableLikeClause will be called after we do know that.
*
* 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 |
@@ -1506,8 +1654,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 *
@@ -2066,10 +2214,10 @@ 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, this queues not-null constraints for each column, if
* needed.
*/
foreach(lc, cxt->ixconstraints)
{
@@ -2143,9 +2291,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);
}
@@ -2153,18 +2299,15 @@ 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 create not-null constraints
* for columns that don't already have them.
*/
static IndexStmt *
transformIndexConstraint(Constraint *constraint, CreateStmtContext *cxt)
{
IndexStmt *index;
List *notnullcmds = NIL;
ListCell *lc;
index = makeNode(IndexStmt);
@@ -2384,6 +2527,12 @@ transformIndexConstraint(Constraint *constraint, CreateStmtContext *cxt)
errdetail("Cannot create a primary key or unique constraint using such an index."),
parser_errposition(cxt->pstate, constraint->location)));
/* If a PK, ensure the columns get not null constraints */
if (constraint->contype == CONSTR_PRIMARY)
cxt->nnconstraints =
lappend(cxt->nnconstraints,
makeNotNullConstraint(makeString(attname)));
constraint->keys = lappend(constraint->keys, makeString(attname));
}
else
@@ -2422,7 +2571,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. For WITHOUT OVERLAPS constraints, we
* also make sure they are not-null. For WITHOUT OVERLAPS constraints, we
* make sure the last part is a range or multirange.
*/
else
@@ -2431,7 +2580,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;
@@ -2453,24 +2601,51 @@ transformIndexConstraint(Constraint *constraint, CreateStmtContext *cxt)
if (found)
{
/*
* column is defined in the new table. For PRIMARY KEY, we
* can apply the not-null constraint cheaply here ... unless
* the column is marked is_from_type, in which case marking it
* here would be ineffective (see MergeAttributes).
* column is defined in the new table. For CREATE TABLE with
* a PRIMARY KEY, we can apply the not-null constraint cheaply
* here. If the not-null constraint already exists, we can
* (albeit not so cheaply) verify that it's not a NO INHERIT
* constraint.
*
* Note that ALTER TABLE never needs either check, because
* those constraints have already been added by
* ATPrepAddPrimaryKey.
*/
if (constraint->contype == CONSTR_PRIMARY &&
!column->is_from_type)
!cxt->isalter)
{
column->is_not_null = true;
forced_not_null = true;
if (column->is_not_null)
{
foreach_node(Constraint, nn, cxt->nnconstraints)
{
if (strcmp(strVal(linitial(nn->keys)), key) == 0)
{
if (nn->is_no_inherit)
ereport(ERROR,
errcode(ERRCODE_SYNTAX_ERROR),
errmsg("conflicting NO INHERIT declaration for not-null constraint on column \"%s\"",
key));
break;
}
}
}
else
{
column->is_not_null = true;
cxt->nnconstraints =
lappend(cxt->nnconstraints,
makeNotNullConstraint(makeString(key)));
}
}
else if (constraint->contype == CONSTR_PRIMARY)
Assert(column->is_not_null);
}
else if (SystemAttributeByName(key) != NULL)
{
/*
* column will be a system column in the new table, so accept
* it. System columns can't ever be null, so no need to worry
* about PRIMARY/not-null constraint.
* about PRIMARY/NOT NULL constraint.
*/
found = true;
}
@@ -2507,13 +2682,10 @@ transformIndexConstraint(Constraint *constraint, CreateStmtContext *cxt)
found = true;
typid = inhattr->atttypid;
/*
* 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.
*/
if (constraint->contype == CONSTR_PRIMARY)
cxt->nnconstraints =
lappend(cxt->nnconstraints,
makeNotNullConstraint(makeString(pstrdup(inhname))));
break;
}
}
@@ -2610,19 +2782,6 @@ transformIndexConstraint(Constraint *constraint, CreateStmtContext *cxt)
iparam->ordering = SORTBY_DEFAULT;
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)
{
AlterTableCmd *notnullcmd = makeNode(AlterTableCmd);
notnullcmd->subtype = AT_SetNotNull;
notnullcmd->name = pstrdup(key);
notnullcmds = lappend(notnullcmds, notnullcmd);
}
}
if (constraint->without_overlaps)
@@ -2741,22 +2900,6 @@ transformIndexConstraint(Constraint *constraint, CreateStmtContext *cxt)
index->indexIncludingParams = lappend(index->indexIncludingParams, iparam);
}
/*
* If we found anything that requires run-time SET NOT NULL, build a full
* ALTER TABLE command for that and add it to cxt->alist.
*/
if (notnullcmds)
{
AlterTableStmt *alterstmt = makeNode(AlterTableStmt);
alterstmt->relation = copyObject(cxt->relation);
alterstmt->cmds = notnullcmds;
alterstmt->objtype = OBJECT_TABLE;
alterstmt->missing_ok = false;
cxt->alist = lappend(cxt->alist, alterstmt);
}
return index;
}
@@ -3395,6 +3538,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;
@@ -3644,9 +3788,8 @@ transformAlterTableStmt(Oid relid, AlterTableStmt *stmt,
Node *istmt = (Node *) lfirst(l);
/*
* 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.
* We assume here that cxt.alist contains only IndexStmts generated
* from primary key constraints.
*/
if (IsA(istmt, IndexStmt))
{
@@ -3658,30 +3801,31 @@ transformAlterTableStmt(Oid relid, AlterTableStmt *stmt,
newcmd->def = (Node *) idxstmt;
newcmds = lappend(newcmds, newcmd);
}
else if (IsA(istmt, AlterTableStmt))
{
AlterTableStmt *alterstmt = (AlterTableStmt *) istmt;
newcmds = list_concat(newcmds, alterstmt->cmds);
}
else
elog(ERROR, "unexpected stmt type %d", (int) nodeTag(istmt));
}
cxt.alist = NIL;
/* Append any CHECK or FK constraints to the commands list */
foreach(l, cxt.ckconstraints)
/* Append any CHECK, NOT NULL or FK constraints to the commands list */
foreach_node(Constraint, def, cxt.ckconstraints)
{
newcmd = makeNode(AlterTableCmd);
newcmd->subtype = AT_AddConstraint;
newcmd->def = (Node *) lfirst_node(Constraint, l);
newcmd->def = (Node *) def;
newcmds = lappend(newcmds, newcmd);
}
foreach(l, cxt.fkconstraints)
foreach_node(Constraint, def, cxt.nnconstraints)
{
newcmd = makeNode(AlterTableCmd);
newcmd->subtype = AT_AddConstraint;
newcmd->def = (Node *) lfirst_node(Constraint, l);
newcmd->def = (Node *) def;
newcmds = lappend(newcmds, newcmd);
}
foreach_node(Constraint, def, cxt.fkconstraints)
{
newcmd = makeNode(AlterTableCmd);
newcmd->subtype = AT_AddConstraint;
newcmd->def = (Node *) def;
newcmds = lappend(newcmds, newcmd);
}

2
src/backend/replication/logical/relation.c
View File

@@ -855,7 +855,7 @@ GetRelationIdentityOrPK(Relation rel)
idxoid = RelationGetReplicaIndex(rel);
if (!OidIsValid(idxoid))
idxoid = RelationGetPrimaryKeyIndex(rel);
idxoid = RelationGetPrimaryKeyIndex(rel, false);
return idxoid;
}

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

@@ -2516,6 +2516,28 @@ pg_get_constraintdef_worker(Oid constraintId, bool fullCommand,
conForm->connoinherit ? " NO INHERIT" : "");
break;
}
case CONSTRAINT_NOTNULL:
{
if (conForm->conrelid)
{
AttrNumber attnum;
attnum = extractNotNullColumn(tup);
appendStringInfo(&buf, "NOT NULL %s",
quote_identifier(get_attname(conForm->conrelid,
attnum, false)));
if (((Form_pg_constraint) GETSTRUCT(tup))->connoinherit)
appendStringInfoString(&buf, " NO INHERIT");
}
else if (conForm->contypid)
{
/* conkey is null for domain not-null constraints */
appendStringInfoString(&buf, "NOT NULL");
}
break;
}
case CONSTRAINT_TRIGGER:
/*

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

@@ -4817,18 +4817,38 @@ 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 or predicate indexes aren't interesting for either oid
* indexes or replication identity indexes, so don't check them.
* Deferred ones are not useful for replication identity either; but
* we do include them if they are PKs.
*/
if (!index->indisvalid || !index->indisunique ||
!index->indimmediate ||
if (!index->indisunique ||
!heap_attisnull(htup, Anum_pg_index_indpred, NULL))
continue;
/* remember primary key index if any */
if (index->indisprimary)
/*
* Remember primary key index, if any. For regular tables we do this
* only if the index is valid; but for partitioned tables, then we do
* it even if it's invalid.
*
* The reason for returning invalid primary keys for partitioned
* tables is that we need it to prevent drop of not-null constraints
* that may underlie such a primary key, which is only a problem for
* partitioned tables.
*/
if (index->indisprimary &&
(index->indisvalid ||
relation->rd_rel->relkind == RELKIND_PARTITIONED_TABLE))
{
pkeyIndex = index->indexrelid;
pkdeferrable = !index->indimmediate;
}
if (!index->indimmediate)
continue;
if (!index->indisvalid)
continue;
/* remember explicitly chosen replica index */
if (index->indisreplident)
@@ -4952,10 +4972,10 @@ RelationGetStatExtList(Relation relation)
* RelationGetPrimaryKeyIndex -- get OID of the relation's primary key index
*
* Returns InvalidOid if there is no such index, or if the primary key is
* DEFERRABLE.
* DEFERRABLE and the caller isn't OK with that.
*/
Oid
RelationGetPrimaryKeyIndex(Relation relation)
RelationGetPrimaryKeyIndex(Relation relation, bool deferrable_ok)
{
List *ilist;
@@ -4967,7 +4987,11 @@ RelationGetPrimaryKeyIndex(Relation relation)
Assert(relation->rd_indexvalid);
}
return relation->rd_ispkdeferrable ? InvalidOid : relation->rd_pkindex;
if (deferrable_ok)
return relation->rd_pkindex;
else if (relation->rd_ispkdeferrable)
return InvalidOid;
return relation->rd_pkindex;
}
/*