From 8583163960b136a52aea7ab02b5cd8b5a81deeca Mon Sep 17 00:00:00 2001
From: jian he <jian.universality@gmail.com>
Date: Sat, 5 Apr 2025 15:21:09 +0800
Subject: [PATCH v7 2/2] NOT NULL NOT VALID

This also introduces a new column, attnotnullvalid, to the pg_attribute catalog.

* You can only add an invalid NOT NULL constraint by using ALTER TABLE.
  CREATE TABLE with NOT NULL NOT VALID will create a valid NOT NULL constraint.

  The syntax to add an invalid NOT NULL constraint:
  ALTER TABLE table_name ADD CONSTRAINT constraint_name NOT NULL column_name NOT VALID;

* To validate a NOT NULL constraint, you can use either of the following:
  ALTER TABLE ... SET NOT NULL
  ALTER TABLE ... VALIDATE CONSTRAINT constraint_name

* ALTER TABLE ... DROP NOT NULL will not remove an invalid NOT NULL constraint.

* CREATE TABLE LIKE will copy the invalid NOT NULL constraints, but in the new table, they will become valid.

Author: Rushabh Lathia <rushabh.lathia@gmail.com>
Author: Jian He <jian.universality@gmail.com>

Discussion: https://postgr.es/m/CAGPqQf0KitkNack4F5CFkFi-9Dqvp29Ro=EpcWt=4_hs-Rt+bQ@mail.gmail.com
---
 contrib/postgres_fdw/postgres_fdw.c       |  26 ++-
 doc/src/sgml/catalogs.sgml                |  11 +-
 doc/src/sgml/ref/alter_table.sgml         |   8 +-
 src/backend/access/common/tupdesc.c       |  11 +-
 src/backend/bootstrap/bootstrap.c         |   3 +
 src/backend/catalog/genbki.pl             |   3 +
 src/backend/catalog/heap.c                |  19 +-
 src/backend/catalog/pg_constraint.c       |  62 ++++--
 src/backend/commands/tablecmds.c          | 241 +++++++++++++++++++---
 src/backend/executor/execMain.c           |   1 +
 src/backend/jit/llvm/llvmjit_deform.c     |  10 +-
 src/backend/optimizer/util/plancat.c      |   6 +-
 src/backend/parser/gram.y                 |   5 +-
 src/backend/parser/parse_utilcmd.c        |  15 ++
 src/backend/utils/cache/catcache.c        |   1 +
 src/backend/utils/cache/relcache.c        |   3 +-
 src/bin/pg_dump/common.c                  |   4 +
 src/bin/pg_dump/pg_dump.c                 | 203 ++++++++++++++++--
 src/bin/pg_dump/pg_dump.h                 |   5 +
 src/bin/pg_dump/t/002_pg_dump.pl          |  17 ++
 src/bin/psql/describe.c                   |   9 +-
 src/include/access/tupdesc.h              |   9 +-
 src/include/catalog/pg_attribute.h        |   5 +-
 src/include/catalog/pg_constraint.h       |   4 +-
 src/test/regress/expected/alter_table.out |  69 +++++++
 src/test/regress/expected/constraints.out | 230 +++++++++++++++++++++
 src/test/regress/sql/alter_table.sql      |  14 ++
 src/test/regress/sql/constraints.sql      | 164 +++++++++++++++
 28 files changed, 1060 insertions(+), 98 deletions(-)

diff --git a/contrib/postgres_fdw/postgres_fdw.c b/contrib/postgres_fdw/postgres_fdw.c
index b4e0e60928b..60d25c2f0a4 100644
--- a/contrib/postgres_fdw/postgres_fdw.c
+++ b/contrib/postgres_fdw/postgres_fdw.c
@@ -5549,7 +5549,15 @@ postgresImportForeignSchema(ImportForeignSchemaStmt *stmt, Oid serverOid)
 							   "SELECT relname, "
 							   "  attname, "
 							   "  format_type(atttypid, atttypmod), "
-							   "  attnotnull, "
+							   "  attnotnull, ");
+
+		/* NOT VALID NOT NULL columns are supported since Postgres 18 */
+		if (PQserverVersion(conn) >= 180000)
+			appendStringInfoString(&buf, "attnotnullvalid, ");
+		else
+			appendStringInfoString(&buf, "attnotnull AS attnotnullvalid, ");
+
+		appendStringInfoString(&buf,
 							   "  pg_get_expr(adbin, adrelid), ");
 
 		/* Generated columns are supported since Postgres 12 */
@@ -5651,6 +5659,7 @@ postgresImportForeignSchema(ImportForeignSchemaStmt *stmt, Oid serverOid)
 				char	   *attname;
 				char	   *typename;
 				char	   *attnotnull;
+				char	   *attnotnullvalid;
 				char	   *attgenerated;
 				char	   *attdefault;
 				char	   *collname;
@@ -5663,14 +5672,15 @@ postgresImportForeignSchema(ImportForeignSchemaStmt *stmt, Oid serverOid)
 				attname = PQgetvalue(res, i, 1);
 				typename = PQgetvalue(res, i, 2);
 				attnotnull = PQgetvalue(res, i, 3);
-				attdefault = PQgetisnull(res, i, 4) ? NULL :
-					PQgetvalue(res, i, 4);
-				attgenerated = PQgetisnull(res, i, 5) ? NULL :
+				attnotnullvalid = PQgetvalue(res, i, 4);
+				attdefault = PQgetisnull(res, i, 5) ? NULL :
 					PQgetvalue(res, i, 5);
-				collname = PQgetisnull(res, i, 6) ? NULL :
+				attgenerated = PQgetisnull(res, i, 6) ? NULL :
 					PQgetvalue(res, i, 6);
-				collnamespace = PQgetisnull(res, i, 7) ? NULL :
+				collname = PQgetisnull(res, i, 7) ? NULL :
 					PQgetvalue(res, i, 7);
+				collnamespace = PQgetisnull(res, i, 8) ? NULL :
+					PQgetvalue(res, i, 8);
 
 				if (first_item)
 					first_item = false;
@@ -5714,7 +5724,11 @@ postgresImportForeignSchema(ImportForeignSchemaStmt *stmt, Oid serverOid)
 
 				/* Add NOT NULL if needed */
 				if (import_not_null && attnotnull[0] == 't')
+				{
 					appendStringInfoString(&buf, " NOT NULL");
+					if (attnotnullvalid[0] == 'f')
+						appendStringInfoString(&buf, " NOT VALID");
+				}
 			}
 			while (++i < numrows &&
 				   strcmp(PQgetvalue(res, i, 0), tablename) == 0);
diff --git a/doc/src/sgml/catalogs.sgml b/doc/src/sgml/catalogs.sgml
index 45ba9c5118f..3ab7d7b68aa 100644
--- a/doc/src/sgml/catalogs.sgml
+++ b/doc/src/sgml/catalogs.sgml
@@ -1260,7 +1260,16 @@
        <structfield>attnotnull</structfield> <type>bool</type>
       </para>
       <para>
-       This column has a not-null constraint.
+       This column has a (possibly unvalidated) not-null constraint.
+      </para></entry>
+     </row>
+
+     <row>
+      <entry role="catalog_table_entry"><para role="column_definition">
+       <structfield>attnotnullvalid</structfield> <type>bool</type>
+      </para>
+      <para>
+       Whether the not-null constraint, if one exists, has been validated.
       </para></entry>
      </row>
 
diff --git a/doc/src/sgml/ref/alter_table.sgml b/doc/src/sgml/ref/alter_table.sgml
index ece438f0075..a75e75d800d 100644
--- a/doc/src/sgml/ref/alter_table.sgml
+++ b/doc/src/sgml/ref/alter_table.sgml
@@ -243,6 +243,8 @@ WITH ( MODULUS <replaceable class="parameter">numeric_literal</replaceable>, REM
       entire table; however, if a valid <literal>CHECK</literal> constraint is
       found which proves no <literal>NULL</literal> can exist, then the
       table scan is skipped.
+      If a column has an invalid not-null constraint,
+      <literal>SET NOT NULL</literal> validates it.
      </para>
 
      <para>
@@ -458,8 +460,8 @@ WITH ( MODULUS <replaceable class="parameter">numeric_literal</replaceable>, REM
      <para>
       This form adds a new constraint to a table using the same constraint
       syntax as <link linkend="sql-createtable"><command>CREATE TABLE</command></link>, plus the option <literal>NOT
-      VALID</literal>, which is currently only allowed for foreign key
-      and CHECK constraints.
+      VALID</literal>, which is currently only allowed for foreign key,
+      <literal>CHECK</literal> constraints and not-null constraints.
      </para>
 
      <para>
@@ -586,7 +588,7 @@ WITH ( MODULUS <replaceable class="parameter">numeric_literal</replaceable>, REM
     <term><literal>VALIDATE CONSTRAINT</literal></term>
     <listitem>
      <para>
-      This form validates a foreign key or check constraint that was
+      This form validates a foreign key, check, or not-null constraint that was
       previously created as <literal>NOT VALID</literal>, by scanning the
       table to ensure there are no rows for which the constraint is not
       satisfied.  If the constraint is not enforced, an error is thrown.
diff --git a/src/backend/access/common/tupdesc.c b/src/backend/access/common/tupdesc.c
index ed2195f14b2..320de2cbda0 100644
--- a/src/backend/access/common/tupdesc.c
+++ b/src/backend/access/common/tupdesc.c
@@ -74,7 +74,8 @@ populate_compact_attribute_internal(Form_pg_attribute src,
 	dst->atthasmissing = src->atthasmissing;
 	dst->attisdropped = src->attisdropped;
 	dst->attgenerated = (src->attgenerated != '\0');
-	dst->attnotnull = src->attnotnull;
+	dst->attnullability = !src->attnotnull ? ATTNULLABLE_NONE :
+		src->attnotnullvalid ? ATTNULLABLE_VALID : ATTNULLABLE_INVALID;
 
 	switch (src->attalign)
 	{
@@ -252,6 +253,7 @@ CreateTupleDescCopy(TupleDesc tupdesc)
 		Form_pg_attribute att = TupleDescAttr(desc, i);
 
 		att->attnotnull = false;
+		att->attnotnullvalid = false;
 		att->atthasdef = false;
 		att->atthasmissing = false;
 		att->attidentity = '\0';
@@ -298,6 +300,7 @@ CreateTupleDescTruncatedCopy(TupleDesc tupdesc, int natts)
 		Form_pg_attribute att = TupleDescAttr(desc, i);
 
 		att->attnotnull = false;
+		att->attnotnullvalid = false;
 		att->atthasdef = false;
 		att->atthasmissing = false;
 		att->attidentity = '\0';
@@ -418,6 +421,7 @@ TupleDescCopy(TupleDesc dst, TupleDesc src)
 		Form_pg_attribute att = TupleDescAttr(dst, i);
 
 		att->attnotnull = false;
+		att->attnotnullvalid = false;
 		att->atthasdef = false;
 		att->atthasmissing = false;
 		att->attidentity = '\0';
@@ -464,6 +468,7 @@ TupleDescCopyEntry(TupleDesc dst, AttrNumber dstAttno,
 
 	/* since we're not copying constraints or defaults, clear these */
 	dstAtt->attnotnull = false;
+	dstAtt->attnotnullvalid = false;
 	dstAtt->atthasdef = false;
 	dstAtt->atthasmissing = false;
 	dstAtt->attidentity = '\0';
@@ -613,6 +618,8 @@ equalTupleDescs(TupleDesc tupdesc1, TupleDesc tupdesc2)
 			return false;
 		if (attr1->attnotnull != attr2->attnotnull)
 			return false;
+		if (attr1->attnotnullvalid != attr2->attnotnullvalid)
+			return false;
 		if (attr1->atthasdef != attr2->atthasdef)
 			return false;
 		if (attr1->attidentity != attr2->attidentity)
@@ -841,6 +848,7 @@ TupleDescInitEntry(TupleDesc desc,
 	att->attndims = attdim;
 
 	att->attnotnull = false;
+	att->attnotnullvalid = false;
 	att->atthasdef = false;
 	att->atthasmissing = false;
 	att->attidentity = '\0';
@@ -904,6 +912,7 @@ TupleDescInitBuiltinEntry(TupleDesc desc,
 	att->attndims = attdim;
 
 	att->attnotnull = false;
+	att->attnotnullvalid = false;
 	att->atthasdef = false;
 	att->atthasmissing = false;
 	att->attidentity = '\0';
diff --git a/src/backend/bootstrap/bootstrap.c b/src/backend/bootstrap/bootstrap.c
index 6db864892d0..44ec9e58520 100644
--- a/src/backend/bootstrap/bootstrap.c
+++ b/src/backend/bootstrap/bootstrap.c
@@ -615,6 +615,9 @@ DefineAttr(char *name, char *type, int attnum, int nullness)
 				attrtypes[attnum]->attnotnull = true;
 		}
 	}
+
+	/* Not-null constraints on system catalogs are always valid. */
+	attrtypes[attnum]->attnotnullvalid = attrtypes[attnum]->attnotnull;
 }
 
 
diff --git a/src/backend/catalog/genbki.pl b/src/backend/catalog/genbki.pl
index df3231fcd41..bdf55d7dc8d 100644
--- a/src/backend/catalog/genbki.pl
+++ b/src/backend/catalog/genbki.pl
@@ -986,6 +986,9 @@ sub morph_row_for_pgattr
 		$row->{attnotnull} = 'f';
 	}
 
+	# Not-null constraints on system catalogs are always valid.
+	$row->{attnotnullvalid} = $row->{attnotnull};
+
 	Catalog::AddDefaultValues($row, $pgattr_schema, 'pg_attribute');
 	return;
 }
diff --git a/src/backend/catalog/heap.c b/src/backend/catalog/heap.c
index 2b28b05b9cb..6620c1cfa8f 100644
--- a/src/backend/catalog/heap.c
+++ b/src/backend/catalog/heap.c
@@ -151,6 +151,7 @@ static const FormData_pg_attribute a1 = {
 	.attalign = TYPALIGN_SHORT,
 	.attstorage = TYPSTORAGE_PLAIN,
 	.attnotnull = true,
+	.attnotnullvalid = true,
 	.attislocal = true,
 };
 
@@ -164,6 +165,7 @@ static const FormData_pg_attribute a2 = {
 	.attalign = TYPALIGN_INT,
 	.attstorage = TYPSTORAGE_PLAIN,
 	.attnotnull = true,
+	.attnotnullvalid = true,
 	.attislocal = true,
 };
 
@@ -177,6 +179,7 @@ static const FormData_pg_attribute a3 = {
 	.attalign = TYPALIGN_INT,
 	.attstorage = TYPSTORAGE_PLAIN,
 	.attnotnull = true,
+	.attnotnullvalid = true,
 	.attislocal = true,
 };
 
@@ -190,6 +193,7 @@ static const FormData_pg_attribute a4 = {
 	.attalign = TYPALIGN_INT,
 	.attstorage = TYPSTORAGE_PLAIN,
 	.attnotnull = true,
+	.attnotnullvalid = true,
 	.attislocal = true,
 };
 
@@ -203,6 +207,7 @@ static const FormData_pg_attribute a5 = {
 	.attalign = TYPALIGN_INT,
 	.attstorage = TYPSTORAGE_PLAIN,
 	.attnotnull = true,
+	.attnotnullvalid = true,
 	.attislocal = true,
 };
 
@@ -222,6 +227,7 @@ static const FormData_pg_attribute a6 = {
 	.attalign = TYPALIGN_INT,
 	.attstorage = TYPSTORAGE_PLAIN,
 	.attnotnull = true,
+	.attnotnullvalid = true,
 	.attislocal = true,
 };
 
@@ -753,6 +759,7 @@ InsertPgAttributeTuples(Relation pg_attribute_rel,
 		slot[slotCount]->tts_values[Anum_pg_attribute_attstorage - 1] = CharGetDatum(attrs->attstorage);
 		slot[slotCount]->tts_values[Anum_pg_attribute_attcompression - 1] = CharGetDatum(attrs->attcompression);
 		slot[slotCount]->tts_values[Anum_pg_attribute_attnotnull - 1] = BoolGetDatum(attrs->attnotnull);
+		slot[slotCount]->tts_values[Anum_pg_attribute_attnotnullvalid - 1] = BoolGetDatum(attrs->attnotnullvalid);
 		slot[slotCount]->tts_values[Anum_pg_attribute_atthasdef - 1] = BoolGetDatum(attrs->atthasdef);
 		slot[slotCount]->tts_values[Anum_pg_attribute_atthasmissing - 1] = BoolGetDatum(attrs->atthasmissing);
 		slot[slotCount]->tts_values[Anum_pg_attribute_attidentity - 1] = CharGetDatum(attrs->attidentity);
@@ -1714,6 +1721,7 @@ RemoveAttributeById(Oid relid, AttrNumber attnum)
 
 	/* Remove any not-null constraint the column may have */
 	attStruct->attnotnull = false;
+	attStruct->attnotnullvalid = false;
 
 	/* Unset this so no one tries to look up the generation expression */
 	attStruct->attgenerated = '\0';
@@ -2616,12 +2624,17 @@ AddRelationNewConstraints(Relation rel,
 						errmsg("cannot add not-null constraint on system column \"%s\"",
 							   strVal(linitial(cdef->keys))));
 
+			Assert(cdef->initially_valid != cdef->skip_validation);
+
 			/*
 			 * If the column already has a not-null constraint, we don't want
-			 * to add another one; just adjust inheritance status as needed.
+			 * to add another one; adjust inheritance status as needed.  This
+			 * also checks whether the existing constraint matches the
+			 * requested validity.
 			 */
-			if (AdjustNotNullInheritance(RelationGetRelid(rel), colnum,
-										 is_local, cdef->is_no_inherit))
+			if (AdjustNotNullInheritance(rel, colnum, is_local,
+										 cdef->is_no_inherit,
+										 cdef->skip_validation))
 				continue;
 
 			/*
diff --git a/src/backend/catalog/pg_constraint.c b/src/backend/catalog/pg_constraint.c
index b97960d2766..a6745c04584 100644
--- a/src/backend/catalog/pg_constraint.c
+++ b/src/backend/catalog/pg_constraint.c
@@ -576,8 +576,8 @@ ChooseConstraintName(const char *name1, const char *name2,
 
 /*
  * Find and return a copy of the pg_constraint tuple that implements a
- * validated not-null constraint for the given column of the given relation.
- * If no such constraint exists, return NULL.
+ * (possibly not valid) 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.
@@ -606,13 +606,11 @@ findNotNullConstraintAttnum(Oid relid, AttrNumber attnum)
 		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.
+		 * We're looking for a NOTNULL constraint 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)
@@ -630,9 +628,10 @@ findNotNullConstraintAttnum(Oid relid, AttrNumber attnum)
 }
 
 /*
- * 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.
+ * Find and return a copy of the pg_constraint tuple that implements a
+ * (possibly not valid) 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)
@@ -723,19 +722,23 @@ extractNotNullColumn(HeapTuple constrTup)
  *
  * 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.
+ * wants, throw an error about the incompatibility.  If the desired
+ * constraint is valid but the existing constraint is not valid, also
+ * throw an error about that (the opposite case is acceptable).
+ *
+ * If everything checks out, we adjust conislocal/coninhcount and return
+ * true.  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)
+AdjustNotNullInheritance(Relation rel, AttrNumber attnum,
+						 bool is_local, bool is_no_inherit, bool is_notvalid)
 {
 	HeapTuple	tup;
 
-	tup = findNotNullConstraintAttnum(relid, attnum);
+	tup = findNotNullConstraintAttnum(RelationGetRelid(rel), attnum);
 	if (HeapTupleIsValid(tup))
 	{
 		Relation	pg_constraint;
@@ -753,7 +756,18 @@ AdjustNotNullInheritance(Oid relid, AttrNumber attnum,
 			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)));
+						   NameStr(conform->conname), RelationGetRelationName(rel)));
+
+		/*
+		 * Throw an error if the existing constraint is NOT VALID and caller
+		 * wants a valid one.
+		 */
+		if (!is_notvalid && !conform->convalidated)
+			ereport(ERROR,
+					errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+					errmsg("incompatible NOT VALID constraint \"%s\" on relation \"%s\"",
+						   NameStr(conform->conname), RelationGetRelationName(rel)),
+					errhint("You will need to use ALTER TABLE ... VALIDATE CONSTRAINT to validate it."));
 
 		if (!is_local)
 		{
@@ -762,6 +776,16 @@ AdjustNotNullInheritance(Oid relid, AttrNumber attnum,
 				ereport(ERROR,
 						errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
 						errmsg("too many inheritance parents"));
+
+			/*
+			 * If the child already has a valid constraint and we are creating
+			 * an invalid one with same definition on it, then the child's
+			 * constraint will remain valid, but can no longer be marked as
+			 * local.
+			*/
+			if (is_notvalid && conform->convalidated && conform->conenforced)
+				conform->conislocal = false;
+
 			changed = true;
 		}
 		else if (!conform->conislocal)
@@ -832,7 +856,7 @@ RelationGetNotNullConstraints(Oid relid, bool cooked, bool include_noinh)
 			cooked->attnum = colnum;
 			cooked->expr = NULL;
 			cooked->is_enforced = true;
-			cooked->skip_validation = false;
+			cooked->skip_validation = !conForm->convalidated;
 			cooked->is_local = true;
 			cooked->inhcount = 0;
 			cooked->is_no_inherit = conForm->connoinherit;
@@ -852,7 +876,7 @@ RelationGetNotNullConstraints(Oid relid, bool cooked, bool include_noinh)
 			constr->keys = list_make1(makeString(get_attname(relid, colnum,
 															 false)));
 			constr->is_enforced = true;
-			constr->skip_validation = false;
+			constr->skip_validation = !conForm->convalidated;
 			constr->initially_valid = true;
 			constr->is_no_inherit = conForm->connoinherit;
 			notnulls = lappend(notnulls, constr);
diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c
index 4397123398e..fcac48db2ac 100644
--- a/src/backend/commands/tablecmds.c
+++ b/src/backend/commands/tablecmds.c
@@ -435,6 +435,9 @@ static void QueueFKConstraintValidation(List **wqueue, Relation conrel, Relation
 static void QueueCheckConstraintValidation(List **wqueue, Relation conrel, Relation rel,
 										   char *constrName, HeapTuple contuple,
 										   bool recurse, bool recursing, LOCKMODE lockmode);
+static void QueueNNConstraintValidation(List **wqueue, Relation conrel, Relation rel,
+										HeapTuple contuple, bool recurse, bool recursing,
+										LOCKMODE lockmode);
 static int	transformColumnNameList(Oid relId, List *colList,
 									int16 *attnums, Oid *atttypids, Oid *attcollids);
 static int	transformFkeyGetPrimaryKey(Relation pkrel, Oid *indexOid,
@@ -498,7 +501,7 @@ static void add_column_collation_dependency(Oid relid, int32 attnum, Oid collid)
 static ObjectAddress ATExecDropNotNull(Relation rel, const char *colName, bool recurse,
 									   LOCKMODE lockmode);
 static void set_attnotnull(List **wqueue, Relation rel, AttrNumber attnum,
-						   LOCKMODE lockmode);
+						   bool is_valid, bool queue_validation);
 static ObjectAddress ATExecSetNotNull(List **wqueue, Relation rel,
 									  char *constrname, char *colName,
 									  bool recurse, bool recursing,
@@ -1340,7 +1343,7 @@ DefineRelation(CreateStmt *stmt, char relkind, Oid ownerId,
 	nncols = AddRelationNotNullConstraints(rel, stmt->nnconstraints,
 										   old_notnulls);
 	foreach_int(attrnum, nncols)
-		set_attnotnull(NULL, rel, attrnum, NoLock);
+		set_attnotnull(NULL, rel, attrnum, true, false);
 
 	ObjectAddressSet(address, RelationRelationId, relationId);
 
@@ -1424,7 +1427,7 @@ BuildDescForRelation(const List *columns)
 		TupleDescInitEntryCollation(desc, attnum, attcollation);
 
 		/* Fill in additional stuff not handled by TupleDescInitEntry */
-		att->attnotnull = entry->is_not_null;
+		att->attnotnull = att->attnotnullvalid = entry->is_not_null;
 		att->attislocal = entry->is_local;
 		att->attinhcount = entry->inhcount;
 		att->attidentity = entry->identity;
@@ -2738,7 +2741,7 @@ MergeAttributes(List *columns, const List *supers, char relpersistence,
 
 		/*
 		 * Request attnotnull on columns that have a not-null constraint
-		 * that's not marked NO INHERIT.
+		 * that's not marked NO INHERIT (even if not valid).
 		 */
 		nnconstrs = RelationGetNotNullConstraints(RelationGetRelid(relation),
 												  true, false);
@@ -6207,18 +6210,22 @@ ATRewriteTable(AlteredTableInfo *tab, Oid OIDNewHeap)
 	{
 		/*
 		 * If we are rebuilding the tuples OR if we added any new but not
-		 * verified not-null constraints, check all not-null constraints. This
-		 * is a bit of overkill but it minimizes risk of bugs.
+		 * verified not-null constraints, check all valid not-null constraints.
+		 * This is a bit of overkill but it minimizes risk of bugs.
 		 *
 		 * notnull_attrs does *not* collect attribute numbers for not-null
 		 * constraints over virtual generated columns; instead, they are
 		 * collected in notnull_virtual_attrs.
+		 *
+		 * But we don't need check invalid not-null constraint! this is aligned
+		 * with check constraint behavior.
 		 */
 		for (i = 0; i < newTupDesc->natts; i++)
 		{
 			Form_pg_attribute attr = TupleDescAttr(newTupDesc, i);
 
-			if (attr->attnotnull && !attr->attisdropped)
+			if (attr->attnotnull && attr->attnotnullvalid &&
+				!attr->attisdropped)
 			{
 				if (attr->attgenerated != ATTRIBUTE_GENERATED_VIRTUAL)
 					notnull_attrs = lappend_int(notnull_attrs, attr->attnum);
@@ -7788,7 +7795,7 @@ ATExecDropNotNull(Relation rel, const char *colName, bool recurse,
 
 	/*
 	 * Find the constraint that makes this column NOT NULL, and drop it.
-	 * dropconstraint_internal() resets attnotnull.
+	 * dropconstraint_internal() resets attnotnull/attnotnullvalid.
 	 */
 	conTup = findNotNullConstraintAttnum(RelationGetRelid(rel), attnum);
 	if (conTup == NULL)
@@ -7809,19 +7816,23 @@ ATExecDropNotNull(Relation rel, const char *colName, bool recurse,
 }
 
 /*
- * Helper to set pg_attribute.attnotnull if it isn't set, and to tell phase 3
- * to verify it.
+ * set_attnotnull
+ *		Helper to update/validate the pg_attribute status of a not-null
+ *		constraint
  *
- * When called to alter an existing table, 'wqueue' must be given so that we
- * can queue a check that existing tuples pass the constraint.  When called
- * from table creation, 'wqueue' should be passed as NULL.
+ * pg_attribute.attnotnull is set true, if it isn't already.  If is_valid
+ * is true, also set pg_attribute.attnotnullvalid.  If queue_validation is
+ * true, also set up wqueue to validate the constraint.  wqueue may be given
+ * as NULL when validation is not needed (e.g., on table creation).
  */
 static void
 set_attnotnull(List **wqueue, Relation rel, AttrNumber attnum,
-			   LOCKMODE lockmode)
+			   bool is_valid, bool queue_validation)
 {
 	Form_pg_attribute attr;
 
+	Assert(!queue_validation || wqueue);
+
 	CheckAlterTableIsSafe(rel);
 
 	/*
@@ -7832,7 +7843,7 @@ set_attnotnull(List **wqueue, Relation rel, AttrNumber attnum,
 	if (attr->attisdropped)
 		return;
 
-	if (!attr->attnotnull)
+	if (!attr->attnotnull || (is_valid && !attr->attnotnullvalid))
 	{
 		Relation	attr_rel;
 		HeapTuple	tuple;
@@ -7845,15 +7856,17 @@ set_attnotnull(List **wqueue, Relation rel, AttrNumber attnum,
 				 attnum, RelationGetRelid(rel));
 
 		attr = (Form_pg_attribute) GETSTRUCT(tuple);
-		Assert(!attr->attnotnull);
+
 		attr->attnotnull = true;
+		attr->attnotnullvalid = is_valid;
 		CatalogTupleUpdate(attr_rel, &tuple->t_self, tuple);
 
 		/*
 		 * If the nullness isn't already proven by validated constraints, have
 		 * ALTER TABLE phase 3 test for it.
 		 */
-		if (wqueue && !NotNullImpliedByRelConstraints(rel, attr))
+		if (queue_validation && wqueue &&
+			!NotNullImpliedByRelConstraints(rel, attr))
 		{
 			AlteredTableInfo *tab;
 
@@ -7951,6 +7964,15 @@ ATExecSetNotNull(List **wqueue, Relation rel, char *conName, char *colName,
 			conForm->conislocal = true;
 			changed = true;
 		}
+		else if (!conForm->convalidated)
+		{
+			/*
+			 * Flip attnotnull and convalidated, and also validate the
+			 * constraint.
+			 */
+			return ATExecValidateConstraint(wqueue, rel, NameStr(conForm->conname),
+											recurse, recursing, lockmode);
+		}
 
 		if (changed)
 		{
@@ -8013,8 +8035,8 @@ ATExecSetNotNull(List **wqueue, Relation rel, char *conName, char *colName,
 	InvokeObjectPostAlterHook(RelationRelationId,
 							  RelationGetRelid(rel), attnum);
 
-	/* Mark pg_attribute.attnotnull for the column */
-	set_attnotnull(wqueue, rel, attnum, lockmode);
+	/* Mark pg_attribute.attnotnull for the column and queue validation */
+	set_attnotnull(wqueue, rel, attnum, true, true);
 
 	/*
 	 * Recurse to propagate the constraint to children that don't have one.
@@ -9456,12 +9478,44 @@ ATPrepAddPrimaryKey(List **wqueue, Relation rel, AlterTableCmd *cmd,
 		}
 	}
 
-	/* Insert not-null constraints in the queue for the PK columns */
+	/* Verify that columns are not-null, or request that they be made so */
 	foreach(lc, pkconstr->keys)
 	{
 		AlterTableCmd *newcmd;
 		Constraint *nnconstr;
+		HeapTuple	tuple;
 
+		/*
+		 * First check if a suitable constraint exists.  If it does, we don't
+		 * need to request another one.  We do need to bail out if it's not
+		 * valid, though.
+		 */
+		tuple = findNotNullConstraint(RelationGetRelid(rel), strVal(lfirst(lc)));
+		if (tuple != NULL)
+		{
+			Form_pg_constraint conForm = (Form_pg_constraint) GETSTRUCT(tuple);
+
+			/* a NO INHERIT constraint is no good */
+			if (conForm->connoinherit)
+				ereport(ERROR,
+						errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+						errmsg("NO INHERIT not-null constraint is incompatible with primary key"),
+						errhint("You will need to use ALTER TABLE ... ALTER CONSTRAINT ... INHERIT."));
+
+			/* an unvalidated constraint is no good */
+			if (!conForm->convalidated)
+				ereport(ERROR,
+						errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+						errmsg("cannot add primary key because of invalid not-null constraint \"%s\"",
+							   NameStr(conForm->conname)),
+						errhint("You will need to use ALTER TABLE ... VALIDATE CONSTRAINT to validate it."));
+
+			/* All good with this one; don't request another */
+			heap_freetuple(tuple);
+			continue;
+		}
+
+		/* This column is not already not-null, so add it to the queue */
 		nnconstr = makeNotNullConstraint(lfirst(lc));
 
 		newcmd = makeNode(AlterTableCmd);
@@ -9836,11 +9890,15 @@ ATAddCheckNNConstraint(List **wqueue, AlteredTableInfo *tab, Relation rel,
 			constr->conname = ccon->name;
 
 		/*
-		 * If adding a not-null constraint, set the pg_attribute flag and tell
-		 * phase 3 to verify existing rows, if needed.
+		 * If adding a valid not-null constraint, set the pg_attribute flag
+		 * and tell phase 3 to verify existing rows, if needed.  For an
+		 * invalid constraint, just set attnotnull and attnotnullvalid,
+		 * without queueing verification.
 		 */
 		if (constr->contype == CONSTR_NOTNULL)
-			set_attnotnull(wqueue, rel, ccon->attnum, lockmode);
+			set_attnotnull(wqueue, rel, ccon->attnum,
+						   !constr->skip_validation,
+						   !constr->skip_validation);
 
 		ObjectAddressSet(address, ConstraintRelationId, ccon->conoid);
 	}
@@ -12811,10 +12869,11 @@ ATExecValidateConstraint(List **wqueue, Relation rel, char *constrName,
 
 	con = (Form_pg_constraint) GETSTRUCT(tuple);
 	if (con->contype != CONSTRAINT_FOREIGN &&
-		con->contype != CONSTRAINT_CHECK)
+		con->contype != CONSTRAINT_CHECK &&
+		con->contype != CONSTRAINT_NOTNULL)
 		ereport(ERROR,
 				(errcode(ERRCODE_WRONG_OBJECT_TYPE),
-				 errmsg("constraint \"%s\" of relation \"%s\" is not a foreign key or check constraint",
+				 errmsg("constraint \"%s\" of relation \"%s\" is not a foreign key, check, or not-null constraint",
 						constrName, RelationGetRelationName(rel))));
 
 	if (!con->conenforced)
@@ -12833,6 +12892,11 @@ ATExecValidateConstraint(List **wqueue, Relation rel, char *constrName,
 			QueueCheckConstraintValidation(wqueue, conrel, rel, constrName,
 										   tuple, recurse, recursing, lockmode);
 		}
+		else if (con->contype == CONSTRAINT_NOTNULL)
+		{
+			QueueNNConstraintValidation(wqueue, conrel, rel,
+										tuple, recurse, recursing, lockmode);
+		}
 
 		ObjectAddressSet(address, ConstraintRelationId, con->oid);
 	}
@@ -13049,6 +13113,109 @@ QueueCheckConstraintValidation(List **wqueue, Relation conrel, Relation rel,
 	heap_freetuple(copyTuple);
 }
 
+/*
+ * QueueNNConstraintValidation
+ *
+ * Add an entry to the wqueue to validate the given not-null constraint in
+ * Phase 3 and update the convalidated field in the pg_constraint catalog for
+ * the specified relation and all its inheriting children.
+ */
+static void
+QueueNNConstraintValidation(List **wqueue, Relation conrel, Relation rel,
+							HeapTuple contuple, bool recurse, bool recursing,
+							LOCKMODE lockmode)
+{
+	Form_pg_constraint con;
+	AlteredTableInfo *tab;
+	HeapTuple	copyTuple;
+	Form_pg_constraint copy_con;
+	List	   *children = NIL;
+	AttrNumber	attnum;
+	char	   *colname;
+
+	con = (Form_pg_constraint) GETSTRUCT(contuple);
+	Assert(con->contype == CONSTRAINT_NOTNULL);
+
+	attnum = extractNotNullColumn(contuple);
+
+	/*
+	 * If we're recursing, we've already done this for parent, so skip it.
+	 * Also, if the constraint is a NO INHERIT constraint, we shouldn't try to
+	 * look for it in the children.
+	 *
+	 * We recurse before validating on the parent, to reduce risk of
+	 * deadlocks.
+	 */
+	if (!recursing && !con->connoinherit)
+		children = find_all_inheritors(RelationGetRelid(rel), lockmode, NULL);
+
+	colname = get_attname(RelationGetRelid(rel), attnum, false);
+	foreach_oid(childoid, children)
+	{
+		Relation	childrel;
+		HeapTuple	contup;
+		Form_pg_constraint childcon;
+		char	   *conname;
+
+		if (childoid == RelationGetRelid(rel))
+			continue;
+
+		/*
+		 * If we are told not to recurse, there had better not be any child
+		 * tables, because we can't mark the constraint on the parent valid
+		 * unless it is valid for all child tables.
+		 */
+		if (!recurse)
+			ereport(ERROR,
+					errcode(ERRCODE_INVALID_TABLE_DEFINITION),
+					errmsg("constraint must be validated on child tables too"));
+
+		/*
+		 * The column on child might have a different attnum, so search by
+		 * column name.
+		 */
+		contup = findNotNullConstraint(childoid, colname);
+		if (!contup)
+			elog(ERROR, "cache lookup failed for not-null constraint on column \"%s\" of relation \"%s\"",
+						colname, get_rel_name(childoid));
+		childcon = (Form_pg_constraint) GETSTRUCT(contup);
+		if (childcon->convalidated)
+			continue;
+
+		/* find_all_inheritors already got lock */
+		childrel = table_open(childoid, NoLock);
+		conname = pstrdup(NameStr(childcon->conname));
+
+		/* XXX improve ATExecValidateConstraint API to avoid double search */
+		ATExecValidateConstraint(wqueue, childrel, conname,
+								 false, true, lockmode);
+		table_close(childrel, NoLock);
+	}
+
+	/* Set the flags appropriately without queueing another validation */
+	set_attnotnull(NULL, rel, attnum, true, false);
+
+	tab = ATGetQueueEntry(wqueue, rel);
+	tab->verify_new_notnull = true;
+
+	/*
+	 * Invalidate relcache so that others see the new validated constraint.
+	 */
+	CacheInvalidateRelcache(rel);
+
+	/*
+	 * Now update the catalogs, while we have the door open.
+	 */
+	copyTuple = heap_copytuple(contuple);
+	copy_con = (Form_pg_constraint) GETSTRUCT(copyTuple);
+	copy_con->convalidated = true;
+	CatalogTupleUpdate(conrel, &copyTuple->t_self, copyTuple);
+
+	InvokeObjectPostAlterHook(ConstraintRelationId, con->oid, 0);
+
+	heap_freetuple(copyTuple);
+}
+
 /*
  * transformColumnNameList - transform list of column names
  *
@@ -13917,10 +14084,11 @@ dropconstraint_internal(Relation rel, HeapTuple constraintTup, DropBehavior beha
 									   false),
 						   RelationGetRelationName(rel)));
 
-		/* All good -- reset attnotnull if needed */
+		/* All good -- reset attnotnull and attnotnullvalid if needed */
 		if (attForm->attnotnull)
 		{
 			attForm->attnotnull = false;
+			attForm->attnotnullvalid = false;
 			CatalogTupleUpdate(attrel, &atttup->t_self, atttup);
 		}
 
@@ -17271,19 +17439,25 @@ MergeAttributesIntoExisting(Relation child_rel, Relation parent_rel, bool ispart
 			if (parent_att->attgenerated && !child_att->attgenerated)
 				ereport(ERROR,
 						(errcode(ERRCODE_DATATYPE_MISMATCH),
-						 errmsg("column \"%s\" in child table must be a generated column", parent_attname)));
+						 errmsg("column \"%s\" in child table must be a generated column",
+								parent_attname)));
 			if (child_att->attgenerated && !parent_att->attgenerated)
 				ereport(ERROR,
 						(errcode(ERRCODE_DATATYPE_MISMATCH),
-						 errmsg("column \"%s\" in child table must not be a generated column", parent_attname)));
+						 errmsg("column \"%s\" in child table must not be a generated column",
+								parent_attname)));
 
-			if (parent_att->attgenerated && child_att->attgenerated && child_att->attgenerated != parent_att->attgenerated)
+			if (parent_att->attgenerated && child_att->attgenerated &&
+				child_att->attgenerated != parent_att->attgenerated)
 				ereport(ERROR,
 						(errcode(ERRCODE_DATATYPE_MISMATCH),
-						 errmsg("column \"%s\" inherits from generated column of different kind", parent_attname),
+						 errmsg("column \"%s\" inherits from generated column of different kind",
+								parent_attname),
 						 errdetail("Parent column is %s, child column is %s.",
-								   parent_att->attgenerated == ATTRIBUTE_GENERATED_STORED ? "STORED" : "VIRTUAL",
-								   child_att->attgenerated == ATTRIBUTE_GENERATED_STORED ? "STORED" : "VIRTUAL")));
+								   parent_att->attgenerated == ATTRIBUTE_GENERATED_STORED ?
+								   "STORED" : "VIRTUAL",
+								   child_att->attgenerated == ATTRIBUTE_GENERATED_STORED ?
+								   "STORED" : "VIRTUAL")));
 
 			/*
 			 * Regular inheritance children are independent enough not to
@@ -19772,7 +19946,8 @@ PartConstraintImpliedByRelConstraint(Relation scanrel,
 		{
 			Form_pg_attribute att = TupleDescAttr(scanrel->rd_att, i - 1);
 
-			if (att->attnotnull && !att->attisdropped)
+			/* invalid not-null constraint must be ignored */
+			if (att->attnotnull && att->attnotnullvalid && !att->attisdropped)
 			{
 				NullTest   *ntest = makeNode(NullTest);
 
diff --git a/src/backend/executor/execMain.c b/src/backend/executor/execMain.c
index 2da848970be..c8d44e3086b 100644
--- a/src/backend/executor/execMain.c
+++ b/src/backend/executor/execMain.c
@@ -2074,6 +2074,7 @@ ExecConstraints(ResultRelInfo *resultRelInfo,
 	{
 		for (AttrNumber attnum = 1; attnum <= tupdesc->natts; attnum++)
 		{
+			/* FIXME use CompactAttribute */
 			Form_pg_attribute att = TupleDescAttr(tupdesc, attnum - 1);
 
 			if (att->attnotnull && att->attgenerated == ATTRIBUTE_GENERATED_VIRTUAL)
diff --git a/src/backend/jit/llvm/llvmjit_deform.c b/src/backend/jit/llvm/llvmjit_deform.c
index 5d169c7a40b..c562edd094b 100644
--- a/src/backend/jit/llvm/llvmjit_deform.c
+++ b/src/backend/jit/llvm/llvmjit_deform.c
@@ -123,7 +123,7 @@ slot_compile_deform(LLVMJitContext *context, TupleDesc desc,
 		 * combination of attisdropped && attnotnull combination shouldn't
 		 * exist.
 		 */
-		if (att->attnotnull &&
+		if (att->attnullability == ATTNULLABLE_VALID &&
 			!att->atthasmissing &&
 			!att->attisdropped)
 			guaranteed_column_number = attnum;
@@ -438,7 +438,7 @@ slot_compile_deform(LLVMJitContext *context, TupleDesc desc,
 		 * into account, because if they're present the heaptuple's natts
 		 * would have indicated that a slot_getmissingattrs() is needed.
 		 */
-		if (!att->attnotnull)
+		if (att->attnullability != ATTNULLABLE_VALID)
 		{
 			LLVMBasicBlockRef b_ifnotnull;
 			LLVMBasicBlockRef b_ifnull;
@@ -604,7 +604,8 @@ slot_compile_deform(LLVMJitContext *context, TupleDesc desc,
 			known_alignment = -1;
 			attguaranteedalign = false;
 		}
-		else if (att->attnotnull && attguaranteedalign && known_alignment >= 0)
+		else if (att->attnullability == ATTNULLABLE_VALID &&
+				 attguaranteedalign && known_alignment >= 0)
 		{
 			/*
 			 * If the offset to the column was previously known, a NOT NULL &
@@ -614,7 +615,8 @@ slot_compile_deform(LLVMJitContext *context, TupleDesc desc,
 			Assert(att->attlen > 0);
 			known_alignment += att->attlen;
 		}
-		else if (att->attnotnull && (att->attlen % alignto) == 0)
+		else if (att->attnullability == ATTNULLABLE_VALID &&
+				 (att->attlen % alignto) == 0)
 		{
 			/*
 			 * After a NOT NULL fixed-width column with a length that is a
diff --git a/src/backend/optimizer/util/plancat.c b/src/backend/optimizer/util/plancat.c
index 441684a72b1..9b4887fc6a7 100644
--- a/src/backend/optimizer/util/plancat.c
+++ b/src/backend/optimizer/util/plancat.c
@@ -177,7 +177,7 @@ get_relation_info(PlannerInfo *root, Oid relationObjectId, bool inhparent,
 		{
 			CompactAttribute *attr = TupleDescCompactAttr(relation->rd_att, i);
 
-			if (attr->attnotnull)
+			if (attr->attnullability == ATTNULLABLE_VALID)
 			{
 				rel->notnullattnums = bms_add_member(rel->notnullattnums,
 													 i + 1);
@@ -1255,6 +1255,7 @@ get_relation_data_width(Oid relid, int32 *attr_widths)
  * get_relation_constraints
  *
  * Retrieve the applicable constraint expressions of the given relation.
+ * Only constraints that have been validated are considered.
  *
  * Returns a List (possibly empty) of constraint expressions.  Each one
  * has been canonicalized, and its Vars are changed to have the varno
@@ -1355,9 +1356,10 @@ get_relation_constraints(PlannerInfo *root,
 
 			for (i = 1; i <= natts; i++)
 			{
+				/* FIXME use CompactAttribute */
 				Form_pg_attribute att = TupleDescAttr(relation->rd_att, i - 1);
 
-				if (att->attnotnull && !att->attisdropped)
+				if (att->attnotnull && att->attnotnullvalid && !att->attisdropped)
 				{
 					NullTest   *ntest = makeNode(NullTest);
 
diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y
index f1156e2fca3..3c4268b271a 100644
--- a/src/backend/parser/gram.y
+++ b/src/backend/parser/gram.y
@@ -4217,11 +4217,10 @@ ConstraintElem:
 					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, NULL,
+								   NULL, NULL, NULL, &n->skip_validation,
 								   &n->is_no_inherit, yyscanner);
-					n->initially_valid = true;
+					n->initially_valid = !n->skip_validation;
 					$$ = (Node *) n;
 				}
 			| UNIQUE opt_unique_null_treatment '(' columnList opt_without_overlaps ')' opt_c_include opt_definition OptConsTableSpace
diff --git a/src/backend/parser/parse_utilcmd.c b/src/backend/parser/parse_utilcmd.c
index 62015431fdf..d2c9f50202e 100644
--- a/src/backend/parser/parse_utilcmd.c
+++ b/src/backend/parser/parse_utilcmd.c
@@ -329,6 +329,21 @@ transformCreateStmt(CreateStmt *stmt, const char *queryString)
 			cd->is_not_null = true;
 			break;
 		}
+
+		if (!cxt.isforeign && !nn->initially_valid)
+		{
+			nn->initially_valid = true;
+			nn->skip_validation = false;
+
+			/*
+			 * not-null constraint created via CREATE TABLE will always be
+			 * valid.  since there is no data there while CREATE TABLE, make
+			 * it invalid does not make sense
+			 */
+			ereport(WARNING,
+					errcode(ERRCODE_WARNING),
+					errmsg("Ignoring NOT VALID flag for NOT NULL constraint on column \"%s\"", colname));
+		}
 	}
 
 	/*
diff --git a/src/backend/utils/cache/catcache.c b/src/backend/utils/cache/catcache.c
index 9ad7681f155..70c11529b90 100644
--- a/src/backend/utils/cache/catcache.c
+++ b/src/backend/utils/cache/catcache.c
@@ -1142,6 +1142,7 @@ CatalogCacheInitializeCache(CatCache *cache)
 			keytype = attr->atttypid;
 			/* cache key columns should always be NOT NULL */
 			Assert(attr->attnotnull);
+			Assert(attr->attnotnullvalid);
 		}
 		else
 		{
diff --git a/src/backend/utils/cache/relcache.c b/src/backend/utils/cache/relcache.c
index 18a14ae186e..4f065f57479 100644
--- a/src/backend/utils/cache/relcache.c
+++ b/src/backend/utils/cache/relcache.c
@@ -592,7 +592,7 @@ RelationBuildTupleDesc(Relation relation)
 
 		/* Update constraint/default info */
 		if (attp->attnotnull)
-			constr->has_not_null = true;
+			constr->has_not_null = true;	/* invalid included */
 		if (attp->attgenerated == ATTRIBUTE_GENERATED_STORED)
 			constr->has_generated_stored = true;
 		if (attp->attgenerated == ATTRIBUTE_GENERATED_VIRTUAL)
@@ -3573,6 +3573,7 @@ RelationBuildLocalRelation(const char *relname,
 		datt->attidentity = satt->attidentity;
 		datt->attgenerated = satt->attgenerated;
 		datt->attnotnull = satt->attnotnull;
+		datt->attnotnullvalid = satt->attnotnullvalid;
 		has_not_null |= satt->attnotnull;
 		populate_compact_attribute(rel->rd_att, i);
 	}
diff --git a/src/bin/pg_dump/common.c b/src/bin/pg_dump/common.c
index 56b6c368acf..ff6a4eacda0 100644
--- a/src/bin/pg_dump/common.c
+++ b/src/bin/pg_dump/common.c
@@ -546,6 +546,10 @@ flagInhAttrs(Archive *fout, DumpOptions *dopt, TableInfo *tblinfo, int numTables
 						parent->notnull_constrs[inhAttrInd] != NULL)
 						foundNotNull = true;
 
+					if (fout->remoteVersion >= 180000 &&
+						parent->notnull_invalid[inhAttrInd])
+						tbinfo->notnull_parent_invalid[j] = true;
+
 					foundDefault |= (parentDef != NULL &&
 									 strcmp(parentDef->adef_expr, "NULL") != 0 &&
 									 !parent->attgenerated[inhAttrInd]);
diff --git a/src/bin/pg_dump/pg_dump.c b/src/bin/pg_dump/pg_dump.c
index 8e6364d32d7..c5cb2d8800c 100644
--- a/src/bin/pg_dump/pg_dump.c
+++ b/src/bin/pg_dump/pg_dump.c
@@ -350,8 +350,10 @@ static void buildMatViewRefreshDependencies(Archive *fout);
 static void getTableDataFKConstraints(void);
 static void determineNotNullFlags(Archive *fout, PGresult *res, int r,
 								  TableInfo *tbinfo, int j,
-								  int i_notnull_name, int i_notnull_noinherit,
-								  int i_notnull_islocal);
+								  int i_notnull_name, int i_notnull_invalidoid,
+								  int i_notnull_noinherit,
+								  int i_notnull_islocal,
+								  PQExpBuffer *invalidnotnulloids);
 static char *format_function_arguments(const FuncInfo *finfo, const char *funcargs,
 									   bool is_agg);
 static char *format_function_signature(Archive *fout,
@@ -8984,6 +8986,7 @@ getTableAttrs(Archive *fout, TableInfo *tblinfo, int numTables)
 	PQExpBuffer q = createPQExpBuffer();
 	PQExpBuffer tbloids = createPQExpBuffer();
 	PQExpBuffer checkoids = createPQExpBuffer();
+	PQExpBuffer invalidnotnulloids = NULL;
 	PGresult   *res;
 	int			ntups;
 	int			curtblindx;
@@ -9003,6 +9006,7 @@ getTableAttrs(Archive *fout, TableInfo *tblinfo, int numTables)
 	int			i_notnull_name;
 	int			i_notnull_noinherit;
 	int			i_notnull_islocal;
+	int			i_notnull_invalidoid;
 	int			i_attoptions;
 	int			i_attcollation;
 	int			i_attcompression;
@@ -9089,6 +9093,10 @@ getTableAttrs(Archive *fout, TableInfo *tblinfo, int numTables)
 	 * attnotnull (this cues dumpTableSchema to print the NOT NULL clause
 	 * without a name); also, such cases are never NO INHERIT.
 	 *
+	 * For invalid constraints, we need to store their OIDs for processing
+	 * elsewhere, so we bring the pg_constraint.oid value when the constraint
+	 * is invalid, and NULL otherwise.
+	 *
 	 * We track in notnull_islocal whether the constraint was defined directly
 	 * in this table or via an ancestor, for binary upgrade.  flagInhAttrs
 	 * might modify this later for servers older than 18; it's also in charge
@@ -9097,11 +9105,14 @@ getTableAttrs(Archive *fout, TableInfo *tblinfo, int numTables)
 	if (fout->remoteVersion >= 180000)
 		appendPQExpBufferStr(q,
 							 "co.conname AS notnull_name,\n"
+							 "CASE WHEN NOT co.convalidated THEN co.oid "
+							 "ELSE NULL END AS notnull_invalidoid,\n"
 							 "co.connoinherit AS notnull_noinherit,\n"
 							 "co.conislocal AS notnull_islocal,\n");
 	else
 		appendPQExpBufferStr(q,
 							 "CASE WHEN a.attnotnull THEN '' ELSE NULL END AS notnull_name,\n"
+							 "NULL AS notnull_invalidoid,\n"
 							 "false AS notnull_noinherit,\n"
 							 "a.attislocal AS notnull_islocal,\n");
 
@@ -9176,6 +9187,7 @@ getTableAttrs(Archive *fout, TableInfo *tblinfo, int numTables)
 	i_attalign = PQfnumber(res, "attalign");
 	i_attislocal = PQfnumber(res, "attislocal");
 	i_notnull_name = PQfnumber(res, "notnull_name");
+	i_notnull_invalidoid = PQfnumber(res, "notnull_invalidoid");
 	i_notnull_noinherit = PQfnumber(res, "notnull_noinherit");
 	i_notnull_islocal = PQfnumber(res, "notnull_islocal");
 	i_attoptions = PQfnumber(res, "attoptions");
@@ -9243,6 +9255,8 @@ getTableAttrs(Archive *fout, TableInfo *tblinfo, int numTables)
 		tbinfo->attfdwoptions = (char **) pg_malloc(numatts * sizeof(char *));
 		tbinfo->attmissingval = (char **) pg_malloc(numatts * sizeof(char *));
 		tbinfo->notnull_constrs = (char **) pg_malloc(numatts * sizeof(char *));
+		tbinfo->notnull_invalid = (bool *) pg_malloc(numatts * sizeof(bool));
+		tbinfo->notnull_parent_invalid = (bool *) pg_malloc(numatts * sizeof(bool));
 		tbinfo->notnull_noinh = (bool *) pg_malloc(numatts * sizeof(bool));
 		tbinfo->notnull_islocal = (bool *) pg_malloc(numatts * sizeof(bool));
 		tbinfo->attrdefs = (AttrDefInfo **) pg_malloc(numatts * sizeof(AttrDefInfo *));
@@ -9268,12 +9282,17 @@ getTableAttrs(Archive *fout, TableInfo *tblinfo, int numTables)
 			tbinfo->attlen[j] = atoi(PQgetvalue(res, r, i_attlen));
 			tbinfo->attalign[j] = *(PQgetvalue(res, r, i_attalign));
 			tbinfo->attislocal[j] = (PQgetvalue(res, r, i_attislocal)[0] == 't');
+			tbinfo->notnull_parent_invalid[j] = false;	/* only change in flagInhAttrs */
+			tbinfo->notnull_invalid[j] = false;			/* only change in determineNotNullFlags */
 
 			/* Handle not-null constraint name and flags */
 			determineNotNullFlags(fout, res, r,
 								  tbinfo, j,
-								  i_notnull_name, i_notnull_noinherit,
-								  i_notnull_islocal);
+								  i_notnull_name,
+								  i_notnull_invalidoid,
+								  i_notnull_noinherit,
+								  i_notnull_islocal,
+								  &invalidnotnulloids);
 
 			tbinfo->attoptions[j] = pg_strdup(PQgetvalue(res, r, i_attoptions));
 			tbinfo->attcollation[j] = atooid(PQgetvalue(res, r, i_attcollation));
@@ -9294,6 +9313,10 @@ getTableAttrs(Archive *fout, TableInfo *tblinfo, int numTables)
 		}
 	}
 
+	/* If invalidnotnulloids has any data, finalize it */
+	if (invalidnotnulloids != NULL)
+		appendPQExpBufferChar(invalidnotnulloids, '}');
+
 	PQclear(res);
 
 	/*
@@ -9426,6 +9449,103 @@ getTableAttrs(Archive *fout, TableInfo *tblinfo, int numTables)
 		PQclear(res);
 	}
 
+	/*
+	 * Get info about NOT NULL NOT VALID constraints.  This is skipped for a
+	 * data-only dump, as it is only needed for table schemas.
+	 */
+	if (dopt->dumpSchema && invalidnotnulloids)
+	{
+		ConstraintInfo *constrs;
+		int			numConstrs;
+		int			i_tableoid;
+		int			i_oid;
+		int			i_conrelid;
+		int			i_conname;
+		int			i_consrc;
+		int			i_conislocal;
+
+		pg_log_info("finding invalid not null constraints");
+
+		resetPQExpBuffer(q);
+		appendPQExpBuffer(q,
+						  "SELECT c.tableoid, c.oid, conrelid, conname, "
+						  "pg_catalog.pg_get_constraintdef(c.oid) AS consrc, "
+						  "conislocal, convalidated "
+						  "FROM unnest('%s'::pg_catalog.oid[]) AS src(conoid)\n"
+						  "JOIN pg_catalog.pg_constraint c ON (src.conoid = c.oid)\n"
+						  "ORDER BY c.conrelid, c.conname",
+						  invalidnotnulloids->data);
+
+		res = ExecuteSqlQuery(fout, q->data, PGRES_TUPLES_OK);
+
+		numConstrs = PQntuples(res);
+		constrs = (ConstraintInfo *) pg_malloc(numConstrs * sizeof(ConstraintInfo));
+
+		i_tableoid = PQfnumber(res, "tableoid");
+		i_oid = PQfnumber(res, "oid");
+		i_conrelid = PQfnumber(res, "conrelid");
+		i_conname = PQfnumber(res, "conname");
+		i_consrc = PQfnumber(res, "consrc");
+		i_conislocal = PQfnumber(res, "conislocal");
+
+		/* As above, this loop iterates once per table, not once per row */
+		curtblindx = -1;
+		for (int j = 0; j < numConstrs;)
+		{
+			Oid			conrelid = atooid(PQgetvalue(res, j, i_conrelid));
+			TableInfo  *tbinfo = NULL;
+			int			numcons;
+
+			/* Count rows for this table */
+			for (numcons = 1; numcons < numConstrs - j; numcons++)
+				if (atooid(PQgetvalue(res, j + numcons, i_conrelid)) != conrelid)
+					break;
+
+			/*
+			 * Locate the associated TableInfo; we rely on tblinfo[] being in
+			 * OID order.
+			 */
+			while (++curtblindx < numTables)
+			{
+				tbinfo = &tblinfo[curtblindx];
+				if (tbinfo->dobj.catId.oid == conrelid)
+					break;
+			}
+			if (curtblindx >= numTables)
+				pg_fatal("unrecognized table OID %u", conrelid);
+
+			for (int c = 0; c < numcons; c++, j++)
+			{
+				constrs[j].dobj.objType = DO_CONSTRAINT;
+				constrs[j].dobj.catId.tableoid = atooid(PQgetvalue(res, j, i_tableoid));
+				constrs[j].dobj.catId.oid = atooid(PQgetvalue(res, j, i_oid));
+				AssignDumpId(&constrs[j].dobj);
+				constrs[j].dobj.name = pg_strdup(PQgetvalue(res, j, i_conname));
+				constrs[j].dobj.namespace = tbinfo->dobj.namespace;
+				constrs[j].contable = tbinfo;
+				constrs[j].condomain = NULL;
+				constrs[j].contype = 'n';
+				constrs[j].condef = pg_strdup(PQgetvalue(res, j, i_consrc));
+				constrs[j].confrelid = InvalidOid;
+				constrs[j].conindex = 0;
+				constrs[j].condeferrable = false;
+				constrs[j].condeferred = false;
+				constrs[j].conislocal = (PQgetvalue(res, j, i_conislocal)[0] == 't');
+
+				/*
+				 * All invalid not-null constraints must be dumped separately,
+				 * because CREATE TABLE would not create them as invalid, and
+				 * also because they must be created after potentially
+				 * violating data has been loaded.
+				 */
+				constrs[j].separate = true;
+
+				constrs[j].dobj.dump = tbinfo->dobj.dump;
+			}
+		}
+		PQclear(res);
+	}
+
 	/*
 	 * Get info about table CHECK constraints.  This is skipped for a
 	 * data-only dump, as it is only needed for table schemas.
@@ -9570,18 +9690,23 @@ getTableAttrs(Archive *fout, TableInfo *tblinfo, int numTables)
  *
  * Result row 'r' is for tbinfo's attribute 'j'.
  *
- * There are three possibilities:
+ * There are four possibilities:
  * 1) the column has no not-null constraints. In that case, ->notnull_constrs
  *    (the constraint name) remains NULL.
  * 2) The column has a constraint with no name (this is the case when
  *    constraints come from pre-18 servers).  In this case, ->notnull_constrs
  *    is set to the empty string; dumpTableSchema will print just "NOT NULL".
- * 3) The column has a constraint with a known name; in that case
+ * 3) The column has an invalid not-null constraint.  This must be treated
+ *    as a separate object (because it must be created after the table data
+ *    is loaded).  So we add its OID to invalidnotnulloids for processing
+ *    elsewhere and do nothing further with it here.  We distinguish this
+ *    case because the "notnull_invalidoid" column has been set to a non-NULL
+ *    value, which is the constraint OID.  Valid constraints have a null OID.
+ * 4) The column has a constraint with a known name; in that case
  *    notnull_constrs carries that name and dumpTableSchema will print
  *    "CONSTRAINT the_name NOT NULL".  However, if the name is the default
  *    (table_column_not_null), there's no need to print that name in the dump,
- *    so notnull_constrs is set to the empty string and it behaves as the case
- *    above.
+ *    so notnull_constrs is set to the empty string and it behaves as case 2.
  *
  * In a child table that inherits from a parent already containing NOT NULL
  * constraints and the columns in the child don't have their own NOT NULL
@@ -9593,7 +9718,7 @@ getTableAttrs(Archive *fout, TableInfo *tblinfo, int numTables)
  * Any of these constraints might have the NO INHERIT bit.  If so we set
  * ->notnull_noinh and NO INHERIT will be printed by dumpTableSchema.
  *
- * In case 3 above, the name comparison is a bit of a hack; it actually fails
+ * In case 4 above, the name comparison is a bit of a hack; it actually fails
  * to do the right thing in all but the trivial case.  However, the downside
  * of getting it wrong is simply that the name is printed rather than
  * suppressed, so it's not a big deal.
@@ -9601,11 +9726,42 @@ getTableAttrs(Archive *fout, TableInfo *tblinfo, int numTables)
 static void
 determineNotNullFlags(Archive *fout, PGresult *res, int r,
 					  TableInfo *tbinfo, int j,
-					  int i_notnull_name, int i_notnull_noinherit,
-					  int i_notnull_islocal)
+					  int i_notnull_name,
+					  int i_notnull_invalidoid,
+					  int i_notnull_noinherit,
+					  int i_notnull_islocal,
+					  PQExpBuffer *invalidnotnulloids)
 {
 	DumpOptions *dopt = fout->dopt;
 
+	/*
+	 * If this not-null constraint is not valid, list its OID in
+	 * invalidnotnulloids and do nothing further.  It'll be processed
+	 * elsewhere later.
+	 *
+	 * Because invalid not-null constraints are rare, we don't want to malloc
+	 * invalidnotnulloids until we're sure we're going it need it, which
+	 * happens here.
+	 */
+	if (!PQgetisnull(res, r, i_notnull_invalidoid))
+	{
+		char	   *constroid = PQgetvalue(res, r, i_notnull_invalidoid);
+
+		if (*invalidnotnulloids == NULL)
+		{
+			*invalidnotnulloids = createPQExpBuffer();
+			appendPQExpBufferChar(*invalidnotnulloids, '{');
+			appendPQExpBuffer(*invalidnotnulloids, "%s", constroid);
+		}
+		else
+			appendPQExpBuffer(*invalidnotnulloids, ",%s", constroid);
+
+		/* nothing else to do */
+		tbinfo->notnull_constrs[j] = NULL;
+		tbinfo->notnull_invalid[j] = true;
+		return;
+	}
+
 	/*
 	 * notnull_noinh is straight from the query result. notnull_islocal also,
 	 * though flagInhAttrs may change that one later in versions < 18.
@@ -16831,6 +16987,18 @@ dumpTableSchema(Archive *fout, const TableInfo *tbinfo)
 											  tbinfo->attrdefs[j]->adef_expr);
 					}
 
+					/*
+					 * if parent have invalid not-null, child have valid
+					 * not-null, then we print not null on child too.  later
+					 * parent's invalid not-null will generate a ALTER TABLE ADD
+					 * CONSTRAINT, which will cascade to children, which is
+					 * fine.
+					*/
+					if (!print_notnull &&
+						tbinfo->notnull_constrs[j] != NULL &&
+						tbinfo->notnull_parent_invalid[j])
+						print_notnull = true;
+
 					if (print_notnull)
 					{
 						if (tbinfo->notnull_constrs[j][0] == '\0')
@@ -18152,13 +18320,20 @@ dumpConstraint(Archive *fout, const ConstraintInfo *coninfo)
 									  .createStmt = q->data,
 									  .dropStmt = delq->data));
 	}
-	else if (coninfo->contype == 'c' && tbinfo)
+	else if ((coninfo->contype == 'c' || coninfo->contype == 'n') && tbinfo)
 	{
-		/* CHECK constraint on a table */
+		/* CHECK or invalid not-null constraint on a table */
 
 		/* Ignore if not to be dumped separately, or if it was inherited */
 		if (coninfo->separate && coninfo->conislocal)
 		{
+			const char *keyword;
+
+			if (coninfo->contype == 'c')
+				keyword = "CHECK CONSTRAINT";
+			else
+				keyword = "CONSTRAINT";
+
 			/* not ONLY since we want it to propagate to children */
 			appendPQExpBuffer(q, "ALTER %sTABLE %s\n", foreign,
 							  fmtQualifiedDumpable(tbinfo));
@@ -18178,7 +18353,7 @@ dumpConstraint(Archive *fout, const ConstraintInfo *coninfo)
 							 ARCHIVE_OPTS(.tag = tag,
 										  .namespace = tbinfo->dobj.namespace->dobj.name,
 										  .owner = tbinfo->rolname,
-										  .description = "CHECK CONSTRAINT",
+										  .description = keyword,
 										  .section = SECTION_POST_DATA,
 										  .createStmt = q->data,
 										  .dropStmt = delq->data));
diff --git a/src/bin/pg_dump/pg_dump.h b/src/bin/pg_dump/pg_dump.h
index e6f0f86a459..5f20d54e30f 100644
--- a/src/bin/pg_dump/pg_dump.h
+++ b/src/bin/pg_dump/pg_dump.h
@@ -365,6 +365,9 @@ typedef struct _tableInfo
 									 * there isn't one on this column. If
 									 * empty string, unnamed constraint
 									 * (pre-v17) */
+	bool	   *notnull_invalid;	/* true if NOT NULL NOT VALID */
+	bool	   *notnull_parent_invalid;	/* true if
+										 * parent table NOT NULL constraint is NOT VALID */
 	bool	   *notnull_noinh;	/* NOT NULL is NO INHERIT */
 	bool	   *notnull_islocal;	/* true if NOT NULL has local definition */
 	struct _attrDefInfo **attrdefs; /* DEFAULT expressions */
@@ -498,6 +501,8 @@ typedef struct _evttriggerInfo
  * use a different objType for foreign key constraints, to make it easier
  * to sort them the way we want.
  *
+ * Not-null constraints don't need this, unless they are NOT VALID.
+ *
  * Note: condeferrable and condeferred are currently only valid for
  * unique/primary-key constraints.  Otherwise that info is in condef.
  */
diff --git a/src/bin/pg_dump/t/002_pg_dump.pl b/src/bin/pg_dump/t/002_pg_dump.pl
index 576326daec7..b6c215c06ff 100644
--- a/src/bin/pg_dump/t/002_pg_dump.pl
+++ b/src/bin/pg_dump/t/002_pg_dump.pl
@@ -1118,6 +1118,23 @@ my %tests = (
 		},
 	},
 
+	'CONSTRAINT NOT NULL / INVALID' => {
+		create_sql => 'CREATE TABLE dump_test.test_table_nn (
+							col1 int);
+			ALTER TABLE dump_test.test_table_nn ADD CONSTRAINT nn NOT NULL col1 NOT VALID;',
+		regexp => qr/^
+			\QALTER TABLE dump_test.test_table_nn\E \n^\s+
+			\QADD CONSTRAINT nn NOT NULL col1 NOT VALID;\E
+			/xm,
+		like => {
+			%full_runs, %dump_test_schema_runs, section_post_data => 1,
+		},
+		unlike => {
+			exclude_dump_test_schema => 1,
+			only_dump_measurement => 1,
+		},
+	},
+
 	'CONSTRAINT PRIMARY KEY / WITHOUT OVERLAPS' => {
 		create_sql => 'CREATE TABLE dump_test.test_table_tpk (
 							col1 int4range,
diff --git a/src/bin/psql/describe.c b/src/bin/psql/describe.c
index 8970677ac64..1d08268393e 100644
--- a/src/bin/psql/describe.c
+++ b/src/bin/psql/describe.c
@@ -3106,7 +3106,8 @@ describeOneTableDetails(const char *schemaname,
 		{
 			printfPQExpBuffer(&buf,
 							  "SELECT c.conname, a.attname, c.connoinherit,\n"
-							  "  c.conislocal, c.coninhcount <> 0\n"
+							  "  c.conislocal, c.coninhcount <> 0,\n"
+							  "  c.convalidated\n"
 							  "FROM pg_catalog.pg_constraint c JOIN\n"
 							  "  pg_catalog.pg_attribute a ON\n"
 							  "    (a.attrelid = c.conrelid AND a.attnum = c.conkey[1])\n"
@@ -3129,14 +3130,16 @@ describeOneTableDetails(const char *schemaname,
 			{
 				bool		islocal = PQgetvalue(result, i, 3)[0] == 't';
 				bool		inherited = PQgetvalue(result, i, 4)[0] == 't';
+				bool		validated = PQgetvalue(result, i, 5)[0] == 't';
 
-				printfPQExpBuffer(&buf, "    \"%s\" NOT NULL \"%s\"%s",
+				printfPQExpBuffer(&buf, "    \"%s\" NOT NULL \"%s\"%s%s",
 								  PQgetvalue(result, i, 0),
 								  PQgetvalue(result, i, 1),
 								  PQgetvalue(result, i, 2)[0] == 't' ?
 								  " NO INHERIT" :
 								  islocal && inherited ? _(" (local, inherited)") :
-								  inherited ? _(" (inherited)") : "");
+								  inherited ? _(" (inherited)") : "",
+								  !validated ? " NOT VALID" : "");
 
 				printTableAddFooter(&cont, buf.data);
 			}
diff --git a/src/include/access/tupdesc.h b/src/include/access/tupdesc.h
index 396eeb7a0bb..5523cfcf5aa 100644
--- a/src/include/access/tupdesc.h
+++ b/src/include/access/tupdesc.h
@@ -42,7 +42,7 @@ typedef struct TupleConstr
 	struct AttrMissing *missing;	/* missing attributes values, NULL if none */
 	uint16		num_defval;
 	uint16		num_check;
-	bool		has_not_null;
+	bool		has_not_null;	/* any not-null, including not valid ones */
 	bool		has_generated_stored;
 	bool		has_generated_virtual;
 } TupleConstr;
@@ -76,10 +76,15 @@ typedef struct CompactAttribute
 	bool		atthasmissing;	/* as FormData_pg_attribute.atthasmissing */
 	bool		attisdropped;	/* as FormData_pg_attribute.attisdropped */
 	bool		attgenerated;	/* FormData_pg_attribute.attgenerated != '\0' */
-	bool		attnotnull;		/* as FormData_pg_attribute.attnotnull */
+	char		attnullability; /* attnotnull + attnotnullvalid */
 	uint8		attalignby;		/* alignment requirement in bytes */
 } CompactAttribute;
 
+/* Valid values for CompactAttribute->attnullability */
+#define		ATTNULLABLE_VALID		'v' /* valid constraint exists */
+#define		ATTNULLABLE_INVALID		'i' /* constraint exists, marked invalid */
+#define		ATTNULLABLE_NONE		'f' /* no constraint exists */
+
 /*
  * This struct is passed around within the backend to describe the structure
  * of tuples.  For tuples coming from on-disk relations, the information is
diff --git a/src/include/catalog/pg_attribute.h b/src/include/catalog/pg_attribute.h
index deaa515fe53..9b1236e7a90 100644
--- a/src/include/catalog/pg_attribute.h
+++ b/src/include/catalog/pg_attribute.h
@@ -117,9 +117,12 @@ CATALOG(pg_attribute,1249,AttributeRelationId) BKI_BOOTSTRAP BKI_ROWTYPE_OID(75,
 	 */
 	char		attcompression BKI_DEFAULT('\0');
 
-	/* This flag represents the "NOT NULL" constraint */
+	/* Whether a not-null constraint exists for the column */
 	bool		attnotnull;
 
+	/* Whether the not-null constraint, if it exists, is valid */
+	bool		attnotnullvalid;
+
 	/* Has DEFAULT value or not */
 	bool		atthasdef BKI_DEFAULT(f);
 
diff --git a/src/include/catalog/pg_constraint.h b/src/include/catalog/pg_constraint.h
index 6da164e7e4d..cf27d53e848 100644
--- a/src/include/catalog/pg_constraint.h
+++ b/src/include/catalog/pg_constraint.h
@@ -263,8 +263,8 @@ extern HeapTuple findNotNullConstraintAttnum(Oid relid, AttrNumber attnum);
 extern HeapTuple findNotNullConstraint(Oid relid, const char *colname);
 extern HeapTuple findDomainNotNullConstraint(Oid typid);
 extern AttrNumber extractNotNullColumn(HeapTuple constrTup);
-extern bool AdjustNotNullInheritance(Oid relid, AttrNumber attnum,
-									 bool is_local, bool is_no_inherit);
+extern bool AdjustNotNullInheritance(Relation rel, AttrNumber attnum,
+									 bool is_local, bool is_no_inherit, bool is_notvalid);
 extern List *RelationGetNotNullConstraints(Oid relid, bool cooked,
 										   bool include_noinh);
 
diff --git a/src/test/regress/expected/alter_table.out b/src/test/regress/expected/alter_table.out
index 362f38856d2..b90b9eac643 100644
--- a/src/test/regress/expected/alter_table.out
+++ b/src/test/regress/expected/alter_table.out
@@ -1202,6 +1202,75 @@ alter table atacc1 alter test_a drop not null, alter test_b drop not null;
 alter table atacc1 add constraint atacc1_constr_b_valid check(test_b is not null);
 alter table atacc1 alter test_b set not null, alter test_a set not null;
 drop table atacc1;
+-- not null not valid with partitions
+CREATE TABLE atnnparted (id int, col1 int) PARTITION BY LIST (id);
+ALTER TABLE atnnparted ADD CONSTRAINT dummy_constr NOT NULL id NOT VALID;
+CREATE TABLE atnnpart1 (col1 int, id int);
+ALTER TABLE atnnpart1 ADD CONSTRAINT another_constr NOT NULL id;
+ALTER TABLE atnnpart1 ADD PRIMARY KEY (id);
+ALTER TABLE atnnparted ATTACH PARTITION atnnpart1 FOR VALUES IN ('1');
+\d+ atnnpart*
+                                 Table "public.atnnpart1"
+ Column |  Type   | Collation | Nullable | Default | Storage | Stats target | Description 
+--------+---------+-----------+----------+---------+---------+--------------+-------------
+ col1   | integer |           |          |         | plain   |              | 
+ id     | integer |           | not null |         | plain   |              | 
+Partition of: atnnparted FOR VALUES IN (1)
+Partition constraint: ((id IS NOT NULL) AND (id = 1))
+Indexes:
+    "atnnpart1_pkey" PRIMARY KEY, btree (id)
+Not-null constraints:
+    "another_constr" NOT NULL "id" (inherited)
+
+                 Index "public.atnnpart1_pkey"
+ Column |  Type   | Key? | Definition | Storage | Stats target 
+--------+---------+------+------------+---------+--------------
+ id     | integer | yes  | id         | plain   | 
+primary key, btree, for table "public.atnnpart1"
+
+                          Partitioned table "public.atnnparted"
+ Column |  Type   | Collation | Nullable | Default | Storage | Stats target | Description 
+--------+---------+-----------+----------+---------+---------+--------------+-------------
+ id     | integer |           | not null |         | plain   |              | 
+ col1   | integer |           |          |         | plain   |              | 
+Partition key: LIST (id)
+Not-null constraints:
+    "dummy_constr" NOT NULL "id" NOT VALID
+Partitions: atnnpart1 FOR VALUES IN (1)
+
+BEGIN;
+ALTER TABLE atnnparted VALIDATE CONSTRAINT dummy_constr;
+\d+ atnnpart*
+                                 Table "public.atnnpart1"
+ Column |  Type   | Collation | Nullable | Default | Storage | Stats target | Description 
+--------+---------+-----------+----------+---------+---------+--------------+-------------
+ col1   | integer |           |          |         | plain   |              | 
+ id     | integer |           | not null |         | plain   |              | 
+Partition of: atnnparted FOR VALUES IN (1)
+Partition constraint: ((id IS NOT NULL) AND (id = 1))
+Indexes:
+    "atnnpart1_pkey" PRIMARY KEY, btree (id)
+Not-null constraints:
+    "another_constr" NOT NULL "id" (inherited)
+
+                 Index "public.atnnpart1_pkey"
+ Column |  Type   | Key? | Definition | Storage | Stats target 
+--------+---------+------+------------+---------+--------------
+ id     | integer | yes  | id         | plain   | 
+primary key, btree, for table "public.atnnpart1"
+
+                          Partitioned table "public.atnnparted"
+ Column |  Type   | Collation | Nullable | Default | Storage | Stats target | Description 
+--------+---------+-----------+----------+---------+---------+--------------+-------------
+ id     | integer |           | not null |         | plain   |              | 
+ col1   | integer |           |          |         | plain   |              | 
+Partition key: LIST (id)
+Not-null constraints:
+    "dummy_constr" NOT NULL "id"
+Partitions: atnnpart1 FOR VALUES IN (1)
+
+ROLLBACK;
+-- leave a table in this state for the pg_restore test
 -- test inheritance
 create table parent (a int);
 create table child (b varchar(255)) inherits (parent);
diff --git a/src/test/regress/expected/constraints.out b/src/test/regress/expected/constraints.out
index a719d2f74e9..90da9e4482f 100644
--- a/src/test/regress/expected/constraints.out
+++ b/src/test/regress/expected/constraints.out
@@ -892,6 +892,222 @@ Not-null constraints:
     "foobar" NOT NULL "a"
 
 DROP TABLE notnull_tbl1;
+-- test NOT NULL NOT VALID
+PREPARE get_nnconstraint_info(regclass[]) AS
+SELECT conrelid::regclass, conname, convalidated, coninhcount
+FROM  pg_constraint
+WHERE conrelid = ANY($1)
+ORDER BY 1;
+CREATE TABLE notnull_tbl1 (a int, b int);
+INSERT INTO notnull_tbl1 VALUES (NULL, 1), (300, 3);
+ALTER TABLE notnull_tbl1 ADD CONSTRAINT nn NOT NULL a; -- error
+ERROR:  column "a" of relation "notnull_tbl1" contains null values
+ALTER TABLE notnull_tbl1 ADD CONSTRAINT nn NOT NULL a NOT VALID; -- ok
+\d+ notnull_tbl1
+                               Table "public.notnull_tbl1"
+ Column |  Type   | Collation | Nullable | Default | Storage | Stats target | Description 
+--------+---------+-----------+----------+---------+---------+--------------+-------------
+ a      | integer |           | not null |         | plain   |              | 
+ b      | integer |           |          |         | plain   |              | 
+Not-null constraints:
+    "nn" NOT NULL "a" NOT VALID
+
+--new table not-null will become valid
+CREATE TABLE notnull_tbl1_copy (LIKE notnull_tbl1);
+EXECUTE get_nnconstraint_info('{notnull_tbl1_copy}');
+     conrelid      | conname | convalidated | coninhcount 
+-------------------+---------+--------------+-------------
+ notnull_tbl1_copy | nn      | t            |           0
+(1 row)
+
+-- even an invalid not-null forbids new nulls
+INSERT INTO notnull_tbl1 VALUES (NULL, 4);
+ERROR:  null value in column "a" of relation "notnull_tbl1" violates not-null constraint
+DETAIL:  Failing row contains (null, 4).
+-- can't override
+ALTER TABLE notnull_tbl1 ADD CONSTRAINT nn1 NOT NULL a NOT VALID NO INHERIT;
+ERROR:  cannot change NO INHERIT status of NOT NULL constraint "nn" on relation "notnull_tbl1"
+ALTER TABLE notnull_tbl1 ADD CONSTRAINT nn NOT NULL a;
+ERROR:  incompatible NOT VALID constraint "nn" on relation "notnull_tbl1"
+HINT:  You will need to use ALTER TABLE ... VALIDATE CONSTRAINT to validate it.
+-- This will validate invalid not-null
+ALTER TABLE notnull_tbl1 ALTER a SET NOT NULL;
+ERROR:  column "a" of relation "notnull_tbl1" contains null values
+-- cannot add primary key on column have invalid not-null
+ALTER TABLE notnull_tbl1 ADD PRIMARY KEY (a);
+ERROR:  cannot add primary key because of invalid not-null constraint "nn"
+HINT:  You will need to use ALTER TABLE ... VALIDATE CONSTRAINT to validate it.
+-- Creating a child table should mark the constraint there as valid
+CREATE TABLE notnull_tbl1_child(a int, b int) INHERITS (notnull_tbl1);
+NOTICE:  merging column "a" with inherited definition
+NOTICE:  merging column "b" with inherited definition
+EXECUTE get_nnconstraint_info('{notnull_tbl1_child, notnull_tbl1}');
+      conrelid      | conname | convalidated | coninhcount 
+--------------------+---------+--------------+-------------
+ notnull_tbl1       | nn      | f            |           0
+ notnull_tbl1_child | nn      | t            |           1
+(2 rows)
+
+ALTER TABLE notnull_tbl1 VALIDATE CONSTRAINT nn; --error
+ERROR:  column "a" of relation "notnull_tbl1" contains null values
+UPDATE notnull_tbl1 SET a = 100 WHERE b = 1;
+ALTER TABLE notnull_tbl1 VALIDATE CONSTRAINT nn; --now ok
+--- now we can add primary key
+ALTER TABLE notnull_tbl1 ADD PRIMARY KEY (a);
+DROP TABLE notnull_tbl1, notnull_tbl1_child;
+-- dropping an invalid constraint is possible
+CREATE TABLE notnull_tbl1 (a int, b int);
+ALTER TABLE notnull_tbl1 ADD NOT NULL a NOT VALID,
+	ADD NOT NULL b NOT VALID;
+ALTER TABLE notnull_tbl1 ALTER a DROP NOT NULL;
+ALTER TABLE notnull_tbl1 DROP CONSTRAINT notnull_tbl1_b_not_null;
+DROP TABLE notnull_tbl1;
+-- ALTER .. NO INHERIT works for invalid constraints
+CREATE TABLE notnull_tbl1 (a int);
+CREATE TABLE notnull_tbl1_chld () INHERITS (notnull_tbl1);
+ALTER TABLE notnull_tbl1 ADD CONSTRAINT nntbl1_a NOT NULL a NOT VALID;
+ALTER TABLE notnull_tbl1 ALTER CONSTRAINT nntbl1_a NO INHERIT;
+-- DROP CONSTRAINT recurses correctly on invalid constraints
+ALTER TABLE notnull_tbl1 ALTER CONSTRAINT nntbl1_a INHERIT;
+ALTER TABLE notnull_tbl1 DROP CONSTRAINT nntbl1_a;
+DROP TABLE notnull_tbl1, notnull_tbl1_chld;
+-- parent have valid not null constraint then child table cannot have invalid one
+CREATE TABLE notnull_tbl1 (a int);
+ALTER TABLE notnull_tbl1 ADD CONSTRAINT nn_parent NOT NULL a not valid;
+CREATE TABLE notnull_chld0 (a int, CONSTRAINT nn_chld0 NOT NULL a);
+ALTER TABLE notnull_tbl1 INHERIT notnull_chld0; --error
+ERROR:  constraint "nn_parent" conflicts with NOT VALID constraint on child table "notnull_tbl1"
+ALTER TABLE notnull_chld0 DROP CONSTRAINT nn_chld0;
+ALTER TABLE notnull_chld0 ADD CONSTRAINT nn_chld0 NOT NULL a not valid;
+ALTER TABLE notnull_tbl1 INHERIT notnull_chld0; --now ok
+-- parents and child not-null will all be validated.
+ALTER TABLE notnull_tbl1 VALIDATE CONSTRAINT nn_parent;
+EXECUTE get_nnconstraint_info('{notnull_tbl1, notnull_chld0}');
+   conrelid    |  conname  | convalidated | coninhcount 
+---------------+-----------+--------------+-------------
+ notnull_tbl1  | nn_parent | t            |           1
+ notnull_chld0 | nn_chld0  | f            |           0
+(2 rows)
+
+DROP TABLE notnull_tbl1, notnull_chld0;
+-- Test invalid not null on inheritance table.
+CREATE TABLE inh_parent ();
+CREATE TABLE inh_child (i int) INHERITS (inh_parent);
+CREATE TABLE inh_grandchild () INHERITS (inh_parent, inh_child);
+ALTER TABLE inh_parent ADD COLUMN i int;
+NOTICE:  merging definition of column "i" for child "inh_child"
+NOTICE:  merging definition of column "i" for child "inh_grandchild"
+ALTER TABLE inh_parent ADD CONSTRAINT nn NOT NULL i NOT VALID;
+ALTER TABLE inh_parent ADD CONSTRAINT nn NOT NULL i; --error
+ERROR:  incompatible NOT VALID constraint "nn" on relation "inh_parent"
+HINT:  You will need to use ALTER TABLE ... VALIDATE CONSTRAINT to validate it.
+ALTER TABLE inh_child ADD CONSTRAINT nn1 NOT NULL i; --error
+ERROR:  incompatible NOT VALID constraint "nn" on relation "inh_child"
+HINT:  You will need to use ALTER TABLE ... VALIDATE CONSTRAINT to validate it.
+EXECUTE get_nnconstraint_info('{inh_parent, inh_child, inh_grandchild}'); --all should be invalid.
+    conrelid    | conname | convalidated | coninhcount 
+----------------+---------+--------------+-------------
+ inh_parent     | nn      | f            |           0
+ inh_child      | nn      | f            |           1
+ inh_grandchild | nn      | f            |           2
+(3 rows)
+
+ALTER TABLE inh_parent ALTER i SET NOT NULL; --ok
+EXECUTE get_nnconstraint_info('{inh_parent, inh_child, inh_grandchild}'); --all should be valid now.
+    conrelid    | conname | convalidated | coninhcount 
+----------------+---------+--------------+-------------
+ inh_parent     | nn      | t            |           0
+ inh_child      | nn      | t            |           1
+ inh_grandchild | nn      | t            |           2
+(3 rows)
+
+DROP TABLE inh_parent, inh_child, inh_grandchild;
+-- Verify NOT NULL VALID/NOT VALID with partition table.
+DROP TABLE notnull_tbl1;
+ERROR:  table "notnull_tbl1" does not exist
+CREATE TABLE notnull_tbl1 (a int, b int) PARTITION BY LIST (a);
+ALTER TABLE notnull_tbl1 ADD CONSTRAINT notnull_con NOT NULL a NOT VALID; --ok
+CREATE TABLE notnull_tbl1_1 PARTITION OF notnull_tbl1 FOR VALUES IN (1,2);
+CREATE TABLE notnull_tbl1_2(a int, CONSTRAINT nn2 NOT NULL a, b int);
+ALTER TABLE notnull_tbl1 ATTACH PARTITION notnull_tbl1_2 FOR VALUES IN (3,4);
+CREATE TABLE notnull_tbl1_3(a int, b int);
+INSERT INTO notnull_tbl1_3 values(NULL,1);
+ALTER TABLE notnull_tbl1_3 add CONSTRAINT nn3 NOT NULL a NOT VALID;
+ALTER TABLE notnull_tbl1 ATTACH PARTITION notnull_tbl1_3 FOR VALUES IN (NULL,5);
+EXECUTE get_nnconstraint_info('{notnull_tbl1, notnull_tbl1_1, notnull_tbl1_2, notnull_tbl1_3}');
+    conrelid    |   conname   | convalidated | coninhcount 
+----------------+-------------+--------------+-------------
+ notnull_tbl1   | notnull_con | f            |           0
+ notnull_tbl1_1 | notnull_con | t            |           1
+ notnull_tbl1_2 | nn2         | t            |           1
+ notnull_tbl1_3 | nn3         | f            |           1
+(4 rows)
+
+ALTER TABLE notnull_tbl1 ALTER COLUMN a SET NOT NULL; --error, notnull_tbl1_3 have null values
+ERROR:  column "a" of relation "notnull_tbl1_3" contains null values
+ALTER TABLE notnull_tbl1_3 VALIDATE CONSTRAINT nn3; --error
+ERROR:  column "a" of relation "notnull_tbl1_3" contains null values
+TRUNCATE notnull_tbl1;
+ALTER TABLE notnull_tbl1 ALTER COLUMN a SET NOT NULL; --OK
+EXECUTE get_nnconstraint_info('{notnull_tbl1, notnull_tbl1_1, notnull_tbl1_2, notnull_tbl1_3}');
+    conrelid    |   conname   | convalidated | coninhcount 
+----------------+-------------+--------------+-------------
+ notnull_tbl1   | notnull_con | t            |           0
+ notnull_tbl1_1 | notnull_con | t            |           1
+ notnull_tbl1_2 | nn2         | t            |           1
+ notnull_tbl1_3 | nn3         | t            |           1
+(4 rows)
+
+DROP TABLE notnull_tbl1;
+-----partitioned table have not-null, then the partitions can not be NOT NULL NOT VALID.
+CREATE TABLE pp_nn (a int, b int, NOT NULL a) PARTITION BY LIST (a);
+CREATE TABLE pp_nn_1(a int, b int);
+ALTER TABLE pp_nn_1 ADD CONSTRAINT nn1 NOT NULL a NOT VALID;
+ALTER TABLE pp_nn ATTACH PARTITION pp_nn_1 FOR VALUES IN (NULL,5); --error
+ERROR:  constraint "nn1" conflicts with NOT VALID constraint on child table "pp_nn_1"
+ALTER TABLE pp_nn_1 VALIDATE CONSTRAINT nn1;
+ALTER TABLE pp_nn ATTACH PARTITION pp_nn_1 FOR VALUES IN (NULL,5); --ok
+DROP TABLE pp_nn;
+-- Create table with NOT NULL INVALID constraint, for pg_upgrade.
+CREATE TABLE notnull_tbl1_upg (a int, b int);
+INSERT INTO notnull_tbl1_upg VALUES (NULL, 1), (NULL, 2), (300, 3);
+ALTER TABLE notnull_tbl1_upg ADD CONSTRAINT nn NOT NULL a NOT VALID;
+-- Inherit test for pg_upgrade
+CREATE TABLE notnull_parent_upg (a int);
+CREATE TABLE notnull_child_upg () INHERITS (notnull_parent_upg);
+ALTER TABLE notnull_child_upg ADD CONSTRAINT nn NOT NULL a;
+ALTER TABLE notnull_parent_upg ADD CONSTRAINT nn NOT NULL a NOT VALID;
+SELECT conrelid::regclass, contype, convalidated, conislocal
+FROM pg_catalog.pg_constraint
+WHERE conrelid in ('notnull_parent_upg'::regclass, 'notnull_child_upg'::regclass)
+ORDER BY 1;
+      conrelid      | contype | convalidated | conislocal 
+--------------------+---------+--------------+------------
+ notnull_parent_upg | n       | f            | t
+ notnull_child_upg  | n       | t            | f
+(2 rows)
+
+-- Partition table test, for pg_upgrade
+CREATE TABLE notnull_part1_upg (a int, b int) PARTITION BY LIST (a);
+ALTER TABLE notnull_part1_upg ADD CONSTRAINT notnull_con NOT NULL a NOT VALID; --ok
+CREATE TABLE notnull_part1_1_upg PARTITION OF notnull_part1_upg FOR VALUES IN (1,2);
+CREATE TABLE notnull_part1_2_upg (a int, CONSTRAINT nn2 NOT NULL a, b int);
+ALTER TABLE notnull_part1_upg ATTACH PARTITION notnull_part1_2_upg FOR VALUES IN (3,4);
+CREATE TABLE notnull_part1_3_upg (a int, b int);
+INSERT INTO notnull_part1_3_upg values(NULL,1);
+ALTER TABLE notnull_part1_3_upg add CONSTRAINT nn3 NOT NULL a NOT VALID;
+ALTER TABLE notnull_part1_upg ATTACH PARTITION notnull_part1_3_upg FOR VALUES IN (NULL,5);
+EXECUTE get_nnconstraint_info('{notnull_part1_upg, notnull_part1_1_upg, notnull_part1_2_upg, notnull_part1_3_upg}');
+      conrelid       |   conname   | convalidated | coninhcount 
+---------------------+-------------+--------------+-------------
+ notnull_part1_upg   | notnull_con | f            |           0
+ notnull_part1_1_upg | notnull_con | t            |           1
+ notnull_part1_2_upg | nn2         | t            |           1
+ notnull_part1_3_upg | nn3         | f            |           1
+(4 rows)
+
+DEALLOCATE get_nnconstraint_info;
+-- end of NOT NULL VALID/NOT VALID --------------------------------
 -- Verify that constraint names and NO INHERIT are properly considered when
 -- multiple constraint are specified, either explicitly or via SERIAL/PK/etc,
 -- and that conflicting cases are rejected.  Mind that table constraints
@@ -1388,3 +1604,17 @@ DROP TABLE constraint_comments_tbl;
 DROP DOMAIN constraint_comments_dom;
 DROP ROLE regress_constraint_comments;
 DROP ROLE regress_constraint_comments_noaccess;
+--sanity check attnotnull and attnotnullvalid.
+select pc.relname, pa.attname, pa.attnum, pa.attnotnull, pa.attnotnullvalid
+from    pg_attribute pa join pg_class pc
+on      pa.attrelid = pc.oid
+where ( (pa.attnotnull and not pa.attnotnullvalid) or
+        (not pa.attnotnull and pa.attnotnullvalid) )
+and     pc.relnamespace in
+        ('pg_catalog'::regnamespace,
+         'pg_toast'::regnamespace,
+         'information_schema'::regnamespace);
+ relname | attname | attnum | attnotnull | attnotnullvalid 
+---------+---------+--------+------------+-----------------
+(0 rows)
+
diff --git a/src/test/regress/sql/alter_table.sql b/src/test/regress/sql/alter_table.sql
index 84e93ef575e..1e6bb2c7818 100644
--- a/src/test/regress/sql/alter_table.sql
+++ b/src/test/regress/sql/alter_table.sql
@@ -910,6 +910,20 @@ alter table atacc1 add constraint atacc1_constr_b_valid check(test_b is not null
 alter table atacc1 alter test_b set not null, alter test_a set not null;
 drop table atacc1;
 
+-- not null not valid with partitions
+CREATE TABLE atnnparted (id int, col1 int) PARTITION BY LIST (id);
+ALTER TABLE atnnparted ADD CONSTRAINT dummy_constr NOT NULL id NOT VALID;
+CREATE TABLE atnnpart1 (col1 int, id int);
+ALTER TABLE atnnpart1 ADD CONSTRAINT another_constr NOT NULL id;
+ALTER TABLE atnnpart1 ADD PRIMARY KEY (id);
+ALTER TABLE atnnparted ATTACH PARTITION atnnpart1 FOR VALUES IN ('1');
+\d+ atnnpart*
+BEGIN;
+ALTER TABLE atnnparted VALIDATE CONSTRAINT dummy_constr;
+\d+ atnnpart*
+ROLLBACK;
+-- leave a table in this state for the pg_restore test
+
 -- test inheritance
 create table parent (a int);
 create table child (b varchar(255)) inherits (parent);
diff --git a/src/test/regress/sql/constraints.sql b/src/test/regress/sql/constraints.sql
index 21ce4177de4..aa28a4a9154 100644
--- a/src/test/regress/sql/constraints.sql
+++ b/src/test/regress/sql/constraints.sql
@@ -640,6 +640,159 @@ ALTER TABLE notnull_tbl1 ADD CONSTRAINT foobar NOT NULL a;
 \d+ notnull_tbl1
 DROP TABLE notnull_tbl1;
 
+-- test NOT NULL NOT VALID
+PREPARE get_nnconstraint_info(regclass[]) AS
+SELECT conrelid::regclass, conname, convalidated, coninhcount
+FROM  pg_constraint
+WHERE conrelid = ANY($1)
+ORDER BY 1;
+
+CREATE TABLE notnull_tbl1 (a int, b int);
+INSERT INTO notnull_tbl1 VALUES (NULL, 1), (300, 3);
+ALTER TABLE notnull_tbl1 ADD CONSTRAINT nn NOT NULL a; -- error
+ALTER TABLE notnull_tbl1 ADD CONSTRAINT nn NOT NULL a NOT VALID; -- ok
+\d+ notnull_tbl1
+
+--new table not-null will become valid
+CREATE TABLE notnull_tbl1_copy (LIKE notnull_tbl1);
+EXECUTE get_nnconstraint_info('{notnull_tbl1_copy}');
+
+-- even an invalid not-null forbids new nulls
+INSERT INTO notnull_tbl1 VALUES (NULL, 4);
+
+-- can't override
+ALTER TABLE notnull_tbl1 ADD CONSTRAINT nn1 NOT NULL a NOT VALID NO INHERIT;
+ALTER TABLE notnull_tbl1 ADD CONSTRAINT nn NOT NULL a;
+
+-- This will validate invalid not-null
+ALTER TABLE notnull_tbl1 ALTER a SET NOT NULL;
+
+-- cannot add primary key on column have invalid not-null
+ALTER TABLE notnull_tbl1 ADD PRIMARY KEY (a);
+
+-- Creating a child table should mark the constraint there as valid
+CREATE TABLE notnull_tbl1_child(a int, b int) INHERITS (notnull_tbl1);
+EXECUTE get_nnconstraint_info('{notnull_tbl1_child, notnull_tbl1}');
+
+ALTER TABLE notnull_tbl1 VALIDATE CONSTRAINT nn; --error
+UPDATE notnull_tbl1 SET a = 100 WHERE b = 1;
+ALTER TABLE notnull_tbl1 VALIDATE CONSTRAINT nn; --now ok
+
+--- now we can add primary key
+ALTER TABLE notnull_tbl1 ADD PRIMARY KEY (a);
+DROP TABLE notnull_tbl1, notnull_tbl1_child;
+
+-- dropping an invalid constraint is possible
+CREATE TABLE notnull_tbl1 (a int, b int);
+ALTER TABLE notnull_tbl1 ADD NOT NULL a NOT VALID,
+	ADD NOT NULL b NOT VALID;
+ALTER TABLE notnull_tbl1 ALTER a DROP NOT NULL;
+ALTER TABLE notnull_tbl1 DROP CONSTRAINT notnull_tbl1_b_not_null;
+DROP TABLE notnull_tbl1;
+
+-- ALTER .. NO INHERIT works for invalid constraints
+CREATE TABLE notnull_tbl1 (a int);
+CREATE TABLE notnull_tbl1_chld () INHERITS (notnull_tbl1);
+ALTER TABLE notnull_tbl1 ADD CONSTRAINT nntbl1_a NOT NULL a NOT VALID;
+ALTER TABLE notnull_tbl1 ALTER CONSTRAINT nntbl1_a NO INHERIT;
+
+-- DROP CONSTRAINT recurses correctly on invalid constraints
+ALTER TABLE notnull_tbl1 ALTER CONSTRAINT nntbl1_a INHERIT;
+ALTER TABLE notnull_tbl1 DROP CONSTRAINT nntbl1_a;
+DROP TABLE notnull_tbl1, notnull_tbl1_chld;
+
+
+-- parent have valid not null constraint then child table cannot have invalid one
+CREATE TABLE notnull_tbl1 (a int);
+ALTER TABLE notnull_tbl1 ADD CONSTRAINT nn_parent NOT NULL a not valid;
+CREATE TABLE notnull_chld0 (a int, CONSTRAINT nn_chld0 NOT NULL a);
+ALTER TABLE notnull_tbl1 INHERIT notnull_chld0; --error
+
+ALTER TABLE notnull_chld0 DROP CONSTRAINT nn_chld0;
+ALTER TABLE notnull_chld0 ADD CONSTRAINT nn_chld0 NOT NULL a not valid;
+ALTER TABLE notnull_tbl1 INHERIT notnull_chld0; --now ok
+
+-- parents and child not-null will all be validated.
+ALTER TABLE notnull_tbl1 VALIDATE CONSTRAINT nn_parent;
+EXECUTE get_nnconstraint_info('{notnull_tbl1, notnull_chld0}');
+DROP TABLE notnull_tbl1, notnull_chld0;
+
+
+-- Test invalid not null on inheritance table.
+CREATE TABLE inh_parent ();
+CREATE TABLE inh_child (i int) INHERITS (inh_parent);
+CREATE TABLE inh_grandchild () INHERITS (inh_parent, inh_child);
+ALTER TABLE inh_parent ADD COLUMN i int;
+ALTER TABLE inh_parent ADD CONSTRAINT nn NOT NULL i NOT VALID;
+ALTER TABLE inh_parent ADD CONSTRAINT nn NOT NULL i; --error
+ALTER TABLE inh_child ADD CONSTRAINT nn1 NOT NULL i; --error
+EXECUTE get_nnconstraint_info('{inh_parent, inh_child, inh_grandchild}'); --all should be invalid.
+ALTER TABLE inh_parent ALTER i SET NOT NULL; --ok
+EXECUTE get_nnconstraint_info('{inh_parent, inh_child, inh_grandchild}'); --all should be valid now.
+DROP TABLE inh_parent, inh_child, inh_grandchild;
+
+
+-- Verify NOT NULL VALID/NOT VALID with partition table.
+DROP TABLE notnull_tbl1;
+CREATE TABLE notnull_tbl1 (a int, b int) PARTITION BY LIST (a);
+ALTER TABLE notnull_tbl1 ADD CONSTRAINT notnull_con NOT NULL a NOT VALID; --ok
+CREATE TABLE notnull_tbl1_1 PARTITION OF notnull_tbl1 FOR VALUES IN (1,2);
+CREATE TABLE notnull_tbl1_2(a int, CONSTRAINT nn2 NOT NULL a, b int);
+ALTER TABLE notnull_tbl1 ATTACH PARTITION notnull_tbl1_2 FOR VALUES IN (3,4);
+
+CREATE TABLE notnull_tbl1_3(a int, b int);
+INSERT INTO notnull_tbl1_3 values(NULL,1);
+ALTER TABLE notnull_tbl1_3 add CONSTRAINT nn3 NOT NULL a NOT VALID;
+ALTER TABLE notnull_tbl1 ATTACH PARTITION notnull_tbl1_3 FOR VALUES IN (NULL,5);
+
+EXECUTE get_nnconstraint_info('{notnull_tbl1, notnull_tbl1_1, notnull_tbl1_2, notnull_tbl1_3}');
+ALTER TABLE notnull_tbl1 ALTER COLUMN a SET NOT NULL; --error, notnull_tbl1_3 have null values
+ALTER TABLE notnull_tbl1_3 VALIDATE CONSTRAINT nn3; --error
+
+TRUNCATE notnull_tbl1;
+ALTER TABLE notnull_tbl1 ALTER COLUMN a SET NOT NULL; --OK
+
+EXECUTE get_nnconstraint_info('{notnull_tbl1, notnull_tbl1_1, notnull_tbl1_2, notnull_tbl1_3}');
+DROP TABLE notnull_tbl1;
+
+-----partitioned table have not-null, then the partitions can not be NOT NULL NOT VALID.
+CREATE TABLE pp_nn (a int, b int, NOT NULL a) PARTITION BY LIST (a);
+CREATE TABLE pp_nn_1(a int, b int);
+ALTER TABLE pp_nn_1 ADD CONSTRAINT nn1 NOT NULL a NOT VALID;
+ALTER TABLE pp_nn ATTACH PARTITION pp_nn_1 FOR VALUES IN (NULL,5); --error
+ALTER TABLE pp_nn_1 VALIDATE CONSTRAINT nn1;
+ALTER TABLE pp_nn ATTACH PARTITION pp_nn_1 FOR VALUES IN (NULL,5); --ok
+DROP TABLE pp_nn;
+
+-- Create table with NOT NULL INVALID constraint, for pg_upgrade.
+CREATE TABLE notnull_tbl1_upg (a int, b int);
+INSERT INTO notnull_tbl1_upg VALUES (NULL, 1), (NULL, 2), (300, 3);
+ALTER TABLE notnull_tbl1_upg ADD CONSTRAINT nn NOT NULL a NOT VALID;
+-- Inherit test for pg_upgrade
+CREATE TABLE notnull_parent_upg (a int);
+CREATE TABLE notnull_child_upg () INHERITS (notnull_parent_upg);
+ALTER TABLE notnull_child_upg ADD CONSTRAINT nn NOT NULL a;
+ALTER TABLE notnull_parent_upg ADD CONSTRAINT nn NOT NULL a NOT VALID;
+SELECT conrelid::regclass, contype, convalidated, conislocal
+FROM pg_catalog.pg_constraint
+WHERE conrelid in ('notnull_parent_upg'::regclass, 'notnull_child_upg'::regclass)
+ORDER BY 1;
+
+-- Partition table test, for pg_upgrade
+CREATE TABLE notnull_part1_upg (a int, b int) PARTITION BY LIST (a);
+ALTER TABLE notnull_part1_upg ADD CONSTRAINT notnull_con NOT NULL a NOT VALID; --ok
+CREATE TABLE notnull_part1_1_upg PARTITION OF notnull_part1_upg FOR VALUES IN (1,2);
+CREATE TABLE notnull_part1_2_upg (a int, CONSTRAINT nn2 NOT NULL a, b int);
+ALTER TABLE notnull_part1_upg ATTACH PARTITION notnull_part1_2_upg FOR VALUES IN (3,4);
+CREATE TABLE notnull_part1_3_upg (a int, b int);
+INSERT INTO notnull_part1_3_upg values(NULL,1);
+ALTER TABLE notnull_part1_3_upg add CONSTRAINT nn3 NOT NULL a NOT VALID;
+ALTER TABLE notnull_part1_upg ATTACH PARTITION notnull_part1_3_upg FOR VALUES IN (NULL,5);
+EXECUTE get_nnconstraint_info('{notnull_part1_upg, notnull_part1_1_upg, notnull_part1_2_upg, notnull_part1_3_upg}');
+DEALLOCATE get_nnconstraint_info;
+-- end of NOT NULL VALID/NOT VALID --------------------------------
+
+
 -- Verify that constraint names and NO INHERIT are properly considered when
 -- multiple constraint are specified, either explicitly or via SERIAL/PK/etc,
 -- and that conflicting cases are rejected.  Mind that table constraints
@@ -836,3 +989,14 @@ DROP DOMAIN constraint_comments_dom;
 
 DROP ROLE regress_constraint_comments;
 DROP ROLE regress_constraint_comments_noaccess;
+
+--sanity check attnotnull and attnotnullvalid.
+select pc.relname, pa.attname, pa.attnum, pa.attnotnull, pa.attnotnullvalid
+from    pg_attribute pa join pg_class pc
+on      pa.attrelid = pc.oid
+where ( (pa.attnotnull and not pa.attnotnullvalid) or
+        (not pa.attnotnull and pa.attnotnullvalid) )
+and     pc.relnamespace in
+        ('pg_catalog'::regnamespace,
+         'pg_toast'::regnamespace,
+         'information_schema'::regnamespace);
-- 
2.34.1

