v4-0001-NOT-NULL-NOT-VALID.patch

From 4c60297c019bebfd5cdcfbfdde3b4f57e168626c Mon Sep 17 00:00:00 2001
From: jian he <jian.universality@gmail.com>
Date: Fri, 21 Mar 2025 22:10:42 +0800
Subject: [PATCH v4 1/1] NOT NULL NOT VALID
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

* it will logically be equivalent to CHECK(x IS NOT NULL) NOT VALID.
* it can only be added using ALTER TABLE. entry point is
 ATAddCheckNNConstraint. when we add a constraint, it will either valid or not valid.
``
    if (constr->contype == CONSTR_NOTNULL)
    {
        if (!constr->skip_validation)
            set_attnotnull(wqueue, rel, ccon->attnum, lockmode);
        else
            set_attinvalidnotnull(rel, ccon->attnum, true);
    }
``
we will only change the state attinvalidnotnull in QueueNNConstraintValidation,
which we will validate the constraint, so attinvalidnotnull should be false.
when we insert a pg_attribute tuple, attinvalidnotnull will always false,
we can even add a assert in InsertPgAttributeTuples: Assert(!attrs->attinvalidnotnull);

* if attnotnull is true, then attinvalidnotnull should be false.
  Conversely, if attinvalidnotnull is true, then attnotnull should be false.
* an invalid not-null cannot be used while adding a primary key.
* if attinvalidnotnull is true, this column can not accept NULL values,
  but the existing column value may contain NULLs, we need to
  VALIDATE the not-null constraint to check if this column exists NULL values or not.
* if partitioned table have valid not-null, then can not attach a not valid to partition tree.
  if partitioned table have not valid not-null, we *can* attach a valid not-null to partition tree.
  (inheritance hierarchy behave the same).

  i tested loaclally, pg_dump will works fine with partitioned table, mainly by tweak
  AdjustNotNullInheritance.
  normal table not null not valid pg_dump also works fine.

discussion: https://postgr.es/m/CAGPqQf0KitkNack4F5CFkFi-9Dqvp29Ro=EpcWt=4_hs-Rt+bQ@mail.gmail.com
---
 doc/src/sgml/catalogs.sgml                |  10 +
 doc/src/sgml/ref/alter_table.sgml         |   2 +
 src/backend/access/common/tupdesc.c       |  11 ++
 src/backend/bootstrap/bootstrap.c         |   1 +
 src/backend/catalog/heap.c                |  11 +-
 src/backend/catalog/pg_constraint.c       |  57 +++++-
 src/backend/commands/tablecmds.c          | 214 ++++++++++++++++++++--
 src/backend/executor/execMain.c           | 130 ++++++++-----
 src/backend/parser/gram.y                 |   4 +-
 src/backend/parser/parse_utilcmd.c        |  14 +-
 src/backend/utils/cache/relcache.c        |   3 +
 src/bin/pg_dump/pg_dump.c                 | 155 +++++++++++++++-
 src/bin/pg_dump/pg_dump.h                 |   1 +
 src/bin/psql/describe.c                   |   9 +-
 src/include/access/tupdesc.h              |   1 +
 src/include/catalog/pg_attribute.h        |   3 +
 src/include/catalog/pg_constraint.h       |  10 +-
 src/test/regress/expected/constraints.out | 169 +++++++++++++++++
 src/test/regress/sql/constraints.sql      | 114 ++++++++++++
 19 files changed, 827 insertions(+), 92 deletions(-)

diff --git a/doc/src/sgml/catalogs.sgml b/doc/src/sgml/catalogs.sgml
index fb050635551..299d2a46f4e 100644
--- a/doc/src/sgml/catalogs.sgml
+++ b/doc/src/sgml/catalogs.sgml
@@ -1264,6 +1264,16 @@
       </para></entry>
      </row>
 
+     <row>
+      <entry role="catalog_table_entry"><para role="column_definition">
+       <structfield>attinvalidnotnull</structfield> <type>bool</type>
+      </para>
+      <para>
+       This column has a attinvalidnotnull not-null constraint.
+       If <structfield>attnotnull</structfield> is true, this has to be false.
+      </para></entry>
+     </row>
+
      <row>
       <entry role="catalog_table_entry"><para role="column_definition">
        <structfield>atthasdef</structfield> <type>bool</type>
diff --git a/doc/src/sgml/ref/alter_table.sgml b/doc/src/sgml/ref/alter_table.sgml
index 4f15b89a98f..75d90654275 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 is marked as <literal>NOT NULL NOT VALID</literal>,
+      <literal>SET NOT NULL</literal> will change it to a validated <literal>NOT NULL</literal> constraint.
      </para>
 
      <para>
diff --git a/src/backend/access/common/tupdesc.c b/src/backend/access/common/tupdesc.c
index ed2195f14b2..634cfddff23 100644
--- a/src/backend/access/common/tupdesc.c
+++ b/src/backend/access/common/tupdesc.c
@@ -252,6 +252,7 @@ CreateTupleDescCopy(TupleDesc tupdesc)
 		Form_pg_attribute att = TupleDescAttr(desc, i);
 
 		att->attnotnull = false;
+		att->attinvalidnotnull = false;
 		att->atthasdef = false;
 		att->atthasmissing = false;
 		att->attidentity = '\0';
@@ -298,6 +299,7 @@ CreateTupleDescTruncatedCopy(TupleDesc tupdesc, int natts)
 		Form_pg_attribute att = TupleDescAttr(desc, i);
 
 		att->attnotnull = false;
+		att->attinvalidnotnull = false;
 		att->atthasdef = false;
 		att->atthasmissing = false;
 		att->attidentity = '\0';
@@ -341,6 +343,7 @@ CreateTupleDescCopyConstr(TupleDesc tupdesc)
 		TupleConstr *cpy = (TupleConstr *) palloc0(sizeof(TupleConstr));
 
 		cpy->has_not_null = constr->has_not_null;
+		cpy->has_invalid_not_null = constr->has_invalid_not_null;
 		cpy->has_generated_stored = constr->has_generated_stored;
 		cpy->has_generated_virtual = constr->has_generated_virtual;
 
@@ -418,6 +421,7 @@ TupleDescCopy(TupleDesc dst, TupleDesc src)
 		Form_pg_attribute att = TupleDescAttr(dst, i);
 
 		att->attnotnull = false;
+		att->attinvalidnotnull = 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->attinvalidnotnull = 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->attinvalidnotnull != attr2->attinvalidnotnull)
+			return false;
 		if (attr1->atthasdef != attr2->atthasdef)
 			return false;
 		if (attr1->attidentity != attr2->attidentity)
@@ -639,6 +646,8 @@ equalTupleDescs(TupleDesc tupdesc1, TupleDesc tupdesc2)
 			return false;
 		if (constr1->has_not_null != constr2->has_not_null)
 			return false;
+		if (constr1->has_invalid_not_null != constr2->has_invalid_not_null)
+			return false;
 		if (constr1->has_generated_stored != constr2->has_generated_stored)
 			return false;
 		if (constr1->has_generated_virtual != constr2->has_generated_virtual)
@@ -841,6 +850,7 @@ TupleDescInitEntry(TupleDesc desc,
 	att->attndims = attdim;
 
 	att->attnotnull = false;
+	att->attinvalidnotnull = false;
 	att->atthasdef = false;
 	att->atthasmissing = false;
 	att->attidentity = '\0';
@@ -904,6 +914,7 @@ TupleDescInitBuiltinEntry(TupleDesc desc,
 	att->attndims = attdim;
 
 	att->attnotnull = false;
+	att->attinvalidnotnull = 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..9077ed46d33 100644
--- a/src/backend/bootstrap/bootstrap.c
+++ b/src/backend/bootstrap/bootstrap.c
@@ -582,6 +582,7 @@ DefineAttr(char *name, char *type, int attnum, int nullness)
 
 	attrtypes[attnum]->atttypmod = -1;
 	attrtypes[attnum]->attislocal = true;
+	attrtypes[attnum]->attinvalidnotnull = false;
 
 	if (nullness == BOOTCOL_NULL_FORCE_NOT_NULL)
 	{
diff --git a/src/backend/catalog/heap.c b/src/backend/catalog/heap.c
index bd3554c0bfd..9b2315a27e7 100644
--- a/src/backend/catalog/heap.c
+++ b/src/backend/catalog/heap.c
@@ -753,6 +753,8 @@ 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_attinvalidnotnull - 1] = BoolGetDatum(attrs->attinvalidnotnull);
+		Assert(!attrs->attinvalidnotnull); /*test demo only*/
 		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);
@@ -2621,12 +2623,17 @@ AddRelationNewConstraints(Relation rel,
 						errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
 						errmsg("not-null constraints are not supported on virtual generated columns"));
 
+			if (cdef->initially_valid)
+				Assert(!cdef->skip_validation);
+			else
+				Assert(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.
 			 */
-			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 ac80652baf2..9b4d0427877 100644
--- a/src/backend/catalog/pg_constraint.c
+++ b/src/backend/catalog/pg_constraint.c
@@ -576,13 +576,14 @@ 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.
+ * if include_invalid is true, it may return an invalid not-null tuple.
  *
  * 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)
+findNotNullConstraintAttnum(Oid relid, AttrNumber attnum, bool include_invalid)
 {
 	Relation	pg_constraint;
 	HeapTuple	conTup,
@@ -609,7 +610,7 @@ findNotNullConstraintAttnum(Oid relid, AttrNumber attnum)
 		 */
 		if (con->contype != CONSTRAINT_NOTNULL)
 			continue;
-		if (!con->convalidated)
+		if (!con->convalidated && !include_invalid)
 			continue;
 
 		conkey = extractNotNullColumn(conTup);
@@ -631,9 +632,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.
+ * if include_invalid is true, it may return an invalid not-null tuple.
  */
 HeapTuple
-findNotNullConstraint(Oid relid, const char *colname)
+findNotNullConstraint(Oid relid, const char *colname, bool include_invalid)
 {
 	AttrNumber	attnum;
 
@@ -641,7 +643,7 @@ findNotNullConstraint(Oid relid, const char *colname)
 	if (attnum <= InvalidAttrNumber)
 		return NULL;
 
-	return findNotNullConstraintAttnum(relid, attnum);
+	return findNotNullConstraintAttnum(relid, attnum, include_invalid);
 }
 
 /*
@@ -728,12 +730,13 @@ extractNotNullColumn(HeapTuple constrTup)
  * 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;
+	Oid	relid = RelationGetRelid(rel);
 
-	tup = findNotNullConstraintAttnum(relid, attnum);
+	tup = findNotNullConstraintAttnum(relid, attnum, true);
 	if (HeapTupleIsValid(tup))
 	{
 		Relation	pg_constraint;
@@ -753,6 +756,36 @@ AdjustNotNullInheritance(Oid relid, AttrNumber attnum,
 					errmsg("cannot change NO INHERIT status of NOT NULL constraint \"%s\" on relation \"%s\"",
 						   NameStr(conform->conname), get_rel_name(relid)));
 
+		/*
+		 * Throw an error if we want change the NOT NULL constraint from NOT
+		 * VALID to VALID.
+		 */
+		if (!is_notvalid && !conform->convalidated)
+			ereport(ERROR,
+					errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+					errmsg("cannot change VALID status of NOT NULL constraint \"%s\" on relation \"%s\"",
+						   NameStr(conform->conname), get_rel_name(relid)),
+					errhint("You may need to use ALTER TABLE VALIDATE CONSTRAINT to validate constraint \"%s\"",
+							NameStr(conform->conname)));
+
+		/* VALID TO NOT VALID seems no usage, so just issue a warning */
+		if (is_notvalid && conform->convalidated)
+		{
+			if (is_local)
+				ereport(WARNING,
+						errmsg("NOT NULL constraint \"%s\" on relation \"%s\" is already valid",
+							NameStr(conform->conname), get_rel_name(relid)));
+			else if (rel->rd_rel->relispartition)
+			{
+				/*
+				* In case of partitions, an inherited not null constraint
+				* is never considered local. See MergeConstraintsIntoExisting also.				
+				*/
+				conform->conislocal = false;
+				changed =true;
+			}
+		}
+
 		if (!is_local)
 		{
 			if (pg_add_s16_overflow(conform->coninhcount, 1,
@@ -788,9 +821,10 @@ AdjustNotNullInheritance(Oid relid, AttrNumber attnum,
  * This is seldom needed, so we just scan pg_constraint each time.
  *
  * 'include_noinh' determines whether to include NO INHERIT constraints or not.
+ * 'include_notvalid' determines whether to include NO VALID constraints or not.
  */
 List *
-RelationGetNotNullConstraints(Oid relid, bool cooked, bool include_noinh)
+RelationGetNotNullConstraints(Oid relid, bool cooked, bool include_noinh, bool include_notvalid)
 {
 	List	   *notnulls = NIL;
 	Relation	constrRel;
@@ -816,6 +850,9 @@ RelationGetNotNullConstraints(Oid relid, bool cooked, bool include_noinh)
 		if (conForm->connoinherit && !include_noinh)
 			continue;
 
+		if (!conForm->convalidated && !include_notvalid)
+			continue;
+
 		colnum = extractNotNullColumn(htup);
 
 		if (cooked)
@@ -830,7 +867,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;
@@ -850,7 +887,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 129c97fdf28..94acf07b63d 100644
--- a/src/backend/commands/tablecmds.c
+++ b/src/backend/commands/tablecmds.c
@@ -410,6 +410,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,
@@ -2713,10 +2716,10 @@ 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. but we will include NOT VALID
 		 */
 		nnconstrs = RelationGetNotNullConstraints(RelationGetRelid(relation),
-												  true, false);
+												  true, false, true);
 		foreach_ptr(CookedConstraint, cc, nnconstrs)
 			nncols = bms_add_member(nncols, cc->attnum);
 
@@ -7711,7 +7714,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.
 	 */
-	conTup = findNotNullConstraintAttnum(RelationGetRelid(rel), attnum);
+	conTup = findNotNullConstraintAttnum(RelationGetRelid(rel), attnum, false);
 	if (conTup == NULL)
 		elog(ERROR, "cache lookup failed for not-null constraint on column \"%s\" of relation \"%s\"",
 			 colName, RelationGetRelationName(rel));
@@ -7729,6 +7732,44 @@ ATExecDropNotNull(Relation rel, const char *colName, bool recurse,
 	return address;
 }
 
+/*
+ * update pg_attribute.attinvalidnotnull
+*/
+static void
+set_attinvalidnotnull(Relation rel, AttrNumber attnum,
+					  bool attinvalidnotnull)
+{
+	Form_pg_attribute attr;
+
+	CheckAlterTableIsSafe(rel);
+
+	attr = TupleDescAttr(RelationGetDescr(rel), attnum - 1);
+	if (attr->attisdropped)
+		return;
+
+	if (attr->attinvalidnotnull != attinvalidnotnull)
+	{
+		Relation	attr_rel;
+		HeapTuple	tuple;
+
+		attr_rel = table_open(AttributeRelationId, RowExclusiveLock);
+
+		tuple = SearchSysCacheCopyAttNum(RelationGetRelid(rel), attnum);
+		if (!HeapTupleIsValid(tuple))
+			elog(ERROR, "cache lookup failed for attribute %d of relation %u",
+				 attnum, RelationGetRelid(rel));
+
+		attr = (Form_pg_attribute) GETSTRUCT(tuple);
+		attr->attinvalidnotnull = attinvalidnotnull;
+
+		CatalogTupleUpdate(attr_rel, &tuple->t_self, tuple);
+		CommandCounterIncrement();
+
+		table_close(attr_rel, RowExclusiveLock);
+		heap_freetuple(tuple);
+	}
+}
+
 /*
  * Helper to set pg_attribute.attnotnull if it isn't set, and to tell phase 3
  * to verify it.
@@ -7845,7 +7886,7 @@ ATExecSetNotNull(List **wqueue, Relation rel, char *conName, char *colName,
 						   colName, RelationGetRelationName(rel))));
 
 	/* See if there's already a constraint */
-	tuple = findNotNullConstraintAttnum(RelationGetRelid(rel), attnum);
+	tuple = findNotNullConstraintAttnum(RelationGetRelid(rel), attnum, true);
 	if (HeapTupleIsValid(tuple))
 	{
 		Form_pg_constraint conForm = (Form_pg_constraint) GETSTRUCT(tuple);
@@ -7880,6 +7921,16 @@ 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)
 		{
@@ -9387,6 +9438,22 @@ ATPrepAddPrimaryKey(List **wqueue, Relation rel, AlterTableCmd *cmd,
 	{
 		AlterTableCmd *newcmd;
 		Constraint *nnconstr;
+		HeapTuple	tuple;
+
+		tuple = findNotNullConstraint(RelationGetRelid(rel), strVal(lfirst(lc)), true);
+		if (tuple != NULL)
+		{
+			Form_pg_constraint conForm = (Form_pg_constraint) GETSTRUCT(tuple);
+			if (!conForm->convalidated)
+				ereport(ERROR,
+						errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+						errmsg("primary key require an validated NOT NULL constraint"),
+						errdetail("Column \"%s\" of table \"%s\" is marked as NOT VALID NOT NULL constraint",
+							      strVal(lfirst(lc)), RelationGetRelationName(rel)),
+						errhint("You may try ALTER TABLE VALIDATE CONSTRAINT to validate constraint \"%s\"",
+								NameStr(conForm->conname)));
+			heap_freetuple(tuple);
+		}
 
 		nnconstr = makeNotNullConstraint(lfirst(lc));
 
@@ -9762,11 +9829,19 @@ ATAddCheckNNConstraint(List **wqueue, AlteredTableInfo *tab, Relation rel,
 			constr->conname = ccon->name;
 
 		/*
-		 * If adding a not-null constraint, set the pg_attribute flag and tell
+		 * If adding a valid not-null constraint, set the pg_attribute flag and tell
 		 * phase 3 to verify existing rows, if needed.
+		 * However if we are adding a invalid not-null constraint, then we only
+		 * need set the pg_attribute.attinvalidnotnull.  phase 3 don't need to
+		 * verify existing rows.
 		 */
 		if (constr->contype == CONSTR_NOTNULL)
-			set_attnotnull(wqueue, rel, ccon->attnum, lockmode);
+		{
+			if (!constr->skip_validation)
+				set_attnotnull(wqueue, rel, ccon->attnum, lockmode);
+			else
+				set_attinvalidnotnull(rel, ccon->attnum, true);
+		}
 
 		ObjectAddressSet(address, ConstraintRelationId, ccon->conoid);
 	}
@@ -12176,7 +12251,7 @@ ATExecAlterConstraintInternal(List **wqueue, ATAlterConstraint *cmdcon,
 				HeapTuple	childtup;
 				Form_pg_constraint childcon;
 
-				childtup = findNotNullConstraint(childoid, colName);
+				childtup = findNotNullConstraint(childoid, colName, false);
 				if (!childtup)
 					elog(ERROR, "cache lookup failed for not-null constraint on column \"%s\" of relation %u",
 						 colName, childoid);
@@ -12367,10 +12442,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 or check constraint or not-null constraint",
 						constrName, RelationGetRelationName(rel))));
 
 	if (!con->conenforced)
@@ -12389,6 +12465,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);
 	}
@@ -12605,6 +12686,116 @@ 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);
+
+	/*
+	 * For constraints that aren't NO INHERIT, we must ensure that we only
+	 * mark the constraint as validated on the parent if it's already
+	 * validated on the children.
+	 *
+	 * If we're recursing, the parent has already done this, so skip it.
+	 *
+	 * 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, true);
+		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);
+	}
+
+	tab = ATGetQueueEntry(wqueue, rel);
+	tab->verify_new_notnull = true;
+
+	set_attnotnull(NULL, rel, attnum, lockmode);
+
+	/*
+	 * 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);
+
+	/*
+	 * set attinvalidnotnull to false. set_attnotnull already tell phase 3 to
+	 * vertify not-null status.
+	 */
+	set_attinvalidnotnull(rel, attnum, false);
+
+	InvokeObjectPostAlterHook(ConstraintRelationId, con->oid, 0);
+
+	heap_freetuple(copyTuple);
+}
+
 /*
  * transformColumnNameList - transform list of column names
  *
@@ -13549,7 +13740,7 @@ dropconstraint_internal(Relation rel, HeapTuple constraintTup, DropBehavior beha
 		 */
 		if (con->contype == CONSTRAINT_NOTNULL)
 		{
-			tuple = findNotNullConstraint(childrelid, colname);
+			tuple = findNotNullConstraint(childrelid, colname, false);
 			if (!HeapTupleIsValid(tuple))
 				elog(ERROR, "cache lookup failed for not-null constraint on column \"%s\" of relation %u",
 					 colname, RelationGetRelid(childrel));
@@ -16813,8 +17004,9 @@ MergeAttributesIntoExisting(Relation child_rel, Relation parent_rel, bool ispart
 			{
 				HeapTuple	contup;
 
+				Assert(!parent_att->attinvalidnotnull);
 				contup = findNotNullConstraintAttnum(RelationGetRelid(parent_rel),
-													 parent_att->attnum);
+													 parent_att->attnum, false);
 				if (HeapTupleIsValid(contup) &&
 					!((Form_pg_constraint) GETSTRUCT(contup))->connoinherit)
 					ereport(ERROR,
diff --git a/src/backend/executor/execMain.c b/src/backend/executor/execMain.c
index e9bd98c7738..06226150f94 100644
--- a/src/backend/executor/execMain.c
+++ b/src/backend/executor/execMain.c
@@ -93,6 +93,9 @@ static bool ExecCheckPermissionsModified(Oid relOid, Oid userid,
 static void ExecCheckXactReadOnly(PlannedStmt *plannedstmt);
 static void EvalPlanQualStart(EPQState *epqstate, Plan *planTree);
 
+static void ReportNotNullViolationError(ResultRelInfo *resultRelInfo,
+										TupleTableSlot *slot,
+										EState *estate, int attrChk);
 /* end of local decls */
 
 
@@ -2071,57 +2074,21 @@ ExecConstraints(ResultRelInfo *resultRelInfo,
 			Form_pg_attribute att = TupleDescAttr(tupdesc, attrChk - 1);
 
 			if (att->attnotnull && slot_attisnull(slot, attrChk))
-			{
-				char	   *val_desc;
-				Relation	orig_rel = rel;
-				TupleDesc	orig_tupdesc = RelationGetDescr(rel);
+				ReportNotNullViolationError(resultRelInfo, slot, estate, attrChk);
+		}
+	}
 
-				/*
-				 * If the tuple has been routed, it's been converted to the
-				 * partition's rowtype, which might differ from the root
-				 * table's.  We must convert it back to the root table's
-				 * rowtype so that val_desc shown error message matches the
-				 * input tuple.
-				 */
-				if (resultRelInfo->ri_RootResultRelInfo)
-				{
-					ResultRelInfo *rootrel = resultRelInfo->ri_RootResultRelInfo;
-					AttrMap    *map;
+	if (constr->has_invalid_not_null)
+	{
+		int			natts = tupdesc->natts;
+		int			attrChk;
 
-					tupdesc = RelationGetDescr(rootrel->ri_RelationDesc);
-					/* a reverse map */
-					map = build_attrmap_by_name_if_req(orig_tupdesc,
-													   tupdesc,
-													   false);
+		for (attrChk = 1; attrChk <= natts; attrChk++)
+		{
+			Form_pg_attribute att = TupleDescAttr(tupdesc, attrChk - 1);
 
-					/*
-					 * Partition-specific slot's tupdesc can't be changed, so
-					 * allocate a new one.
-					 */
-					if (map != NULL)
-						slot = execute_attr_map_slot(map, slot,
-													 MakeTupleTableSlot(tupdesc, &TTSOpsVirtual));
-					modifiedCols = bms_union(ExecGetInsertedCols(rootrel, estate),
-											 ExecGetUpdatedCols(rootrel, estate));
-					rel = rootrel->ri_RelationDesc;
-				}
-				else
-					modifiedCols = bms_union(ExecGetInsertedCols(resultRelInfo, estate),
-											 ExecGetUpdatedCols(resultRelInfo, estate));
-				val_desc = ExecBuildSlotValueDescription(RelationGetRelid(rel),
-														 slot,
-														 tupdesc,
-														 modifiedCols,
-														 64);
-
-				ereport(ERROR,
-						(errcode(ERRCODE_NOT_NULL_VIOLATION),
-						 errmsg("null value in column \"%s\" of relation \"%s\" violates not-null constraint",
-								NameStr(att->attname),
-								RelationGetRelationName(orig_rel)),
-						 val_desc ? errdetail("Failing row contains %s.", val_desc) : 0,
-						 errtablecol(orig_rel, attrChk)));
-			}
+			if (att->attinvalidnotnull && slot_attisnull(slot, attrChk))
+				ReportNotNullViolationError(resultRelInfo, slot, estate, attrChk);
 		}
 	}
 
@@ -2176,6 +2143,73 @@ ExecConstraints(ResultRelInfo *resultRelInfo,
 	}
 }
 
+
+/*
+ * Report a violation of a not-null constraint that was already detected.
+ */
+static void
+ReportNotNullViolationError(ResultRelInfo *resultRelInfo, TupleTableSlot *slot,
+							EState *estate, int attrChk)
+{
+	Bitmapset  *modifiedCols;
+	char	   *val_desc;
+
+	Relation	rel = resultRelInfo->ri_RelationDesc;
+	Relation	orig_rel = rel;
+
+	TupleDesc	tupdesc = RelationGetDescr(rel);
+	TupleDesc	orig_tupdesc = RelationGetDescr(rel);
+	Form_pg_attribute att = TupleDescAttr(tupdesc, attrChk - 1);
+
+	Assert(attrChk > 0);
+
+	/*
+	 * If the tuple has been routed, it's been converted to the partition's
+	 * rowtype, which might differ from the root table's.  We must convert it
+	 * back to the root table's rowtype so that val_desc shown error message
+	 * matches the input tuple.
+	 */
+	if (resultRelInfo->ri_RootResultRelInfo)
+	{
+		ResultRelInfo *rootrel = resultRelInfo->ri_RootResultRelInfo;
+		AttrMap    *map;
+
+		tupdesc = RelationGetDescr(rootrel->ri_RelationDesc);
+		/* a reverse map */
+		map = build_attrmap_by_name_if_req(orig_tupdesc,
+										   tupdesc,
+										   false);
+
+		/*
+		 * Partition-specific slot's tupdesc can't be changed, so allocate a
+		 * new one.
+		 */
+		if (map != NULL)
+			slot = execute_attr_map_slot(map, slot,
+										 MakeTupleTableSlot(tupdesc, &TTSOpsVirtual));
+		modifiedCols = bms_union(ExecGetInsertedCols(rootrel, estate),
+								 ExecGetUpdatedCols(rootrel, estate));
+		rel = rootrel->ri_RelationDesc;
+	}
+	else
+		modifiedCols = bms_union(ExecGetInsertedCols(resultRelInfo, estate),
+								 ExecGetUpdatedCols(resultRelInfo, estate));
+
+	val_desc = ExecBuildSlotValueDescription(RelationGetRelid(rel),
+											 slot,
+											 tupdesc,
+											 modifiedCols,
+											 64);
+	ereport(ERROR,
+			errcode(ERRCODE_NOT_NULL_VIOLATION),
+			errmsg("null value in column \"%s\" of relation \"%s\" violates not-null constraint",
+				   NameStr(att->attname),
+				   RelationGetRelationName(orig_rel)),
+			val_desc ? errdetail("Failing row contains %s.", val_desc) : 0,
+			errtablecol(orig_rel, attrChk));
+}
+
+
 /*
  * ExecWithCheckOptions -- check that tuple satisfies any WITH CHECK OPTIONs
  * of the specified kind.
diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y
index 271ae26cbaf..715eb51300b 100644
--- a/src/backend/parser/gram.y
+++ b/src/backend/parser/gram.y
@@ -4224,9 +4224,9 @@ ConstraintElem:
 					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 896a7f2c59b..59d689640c6 100644
--- a/src/backend/parser/parse_utilcmd.c
+++ b/src/backend/parser/parse_utilcmd.c
@@ -329,6 +329,18 @@ 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;
+			/*
+			 * we do not support NOT NULL NOT VALID for column constraint,
+			 * nn->conname should not be NULL.
+			*/
+			ereport(WARNING,
+					errcode(ERRCODE_WARNING),
+					errmsg("Ignoring NOT VALID flag for NOT NULL constraint \"%s\"", nn->conname));
+		}
 	}
 
 	/*
@@ -1291,7 +1303,7 @@ transformTableLikeClause(CreateStmtContext *cxt, TableLikeClause *table_like_cla
 		List	   *lst;
 
 		lst = RelationGetNotNullConstraints(RelationGetRelid(relation), false,
-											true);
+											true, false);
 		cxt->nnconstraints = list_concat(cxt->nnconstraints, lst);
 	}
 
diff --git a/src/backend/utils/cache/relcache.c b/src/backend/utils/cache/relcache.c
index 9f54a9e72b7..d5cd93e55c3 100644
--- a/src/backend/utils/cache/relcache.c
+++ b/src/backend/utils/cache/relcache.c
@@ -593,6 +593,8 @@ RelationBuildTupleDesc(Relation relation)
 		/* Update constraint/default info */
 		if (attp->attnotnull)
 			constr->has_not_null = true;
+		if (attp->attinvalidnotnull)
+			constr->has_invalid_not_null = true;
 		if (attp->attgenerated == ATTRIBUTE_GENERATED_STORED)
 			constr->has_generated_stored = true;
 		if (attp->attgenerated == ATTRIBUTE_GENERATED_VIRTUAL)
@@ -678,6 +680,7 @@ RelationBuildTupleDesc(Relation relation)
 	 * Set up constraint/default info
 	 */
 	if (constr->has_not_null ||
+		constr->has_invalid_not_null ||
 		constr->has_generated_stored ||
 		constr->has_generated_virtual ||
 		ndef > 0 ||
diff --git a/src/bin/pg_dump/pg_dump.c b/src/bin/pg_dump/pg_dump.c
index 428ed2d60fc..1d67dcac124 100644
--- a/src/bin/pg_dump/pg_dump.c
+++ b/src/bin/pg_dump/pg_dump.c
@@ -8884,6 +8884,7 @@ getTableAttrs(Archive *fout, TableInfo *tblinfo, int numTables)
 	PQExpBuffer q = createPQExpBuffer();
 	PQExpBuffer tbloids = createPQExpBuffer();
 	PQExpBuffer checkoids = createPQExpBuffer();
+	PQExpBuffer invalidnotnulloids = createPQExpBuffer();
 	PGresult   *res;
 	int			ntups;
 	int			curtblindx;
@@ -8903,6 +8904,7 @@ getTableAttrs(Archive *fout, TableInfo *tblinfo, int numTables)
 	int			i_notnull_name;
 	int			i_notnull_noinherit;
 	int			i_notnull_islocal;
+	int			i_notnull_validated;
 	int			i_attoptions;
 	int			i_attcollation;
 	int			i_attcompression;
@@ -8998,12 +9000,14 @@ getTableAttrs(Archive *fout, TableInfo *tblinfo, int numTables)
 		appendPQExpBufferStr(q,
 							 "co.conname AS notnull_name,\n"
 							 "co.connoinherit AS notnull_noinherit,\n"
-							 "co.conislocal AS notnull_islocal,\n");
+							 "co.conislocal AS notnull_islocal,\n"
+							 "co.convalidated as notnull_validated,\n");
 	else
 		appendPQExpBufferStr(q,
 							 "CASE WHEN a.attnotnull THEN '' ELSE NULL END AS notnull_name,\n"
 							 "false AS notnull_noinherit,\n"
-							 "a.attislocal AS notnull_islocal,\n");
+							 "a.attislocal AS notnull_islocal,\n"
+							 "true as notnull_validated\n");
 
 	if (fout->remoteVersion >= 140000)
 		appendPQExpBufferStr(q,
@@ -9078,6 +9082,7 @@ getTableAttrs(Archive *fout, TableInfo *tblinfo, int numTables)
 	i_notnull_name = PQfnumber(res, "notnull_name");
 	i_notnull_noinherit = PQfnumber(res, "notnull_noinherit");
 	i_notnull_islocal = PQfnumber(res, "notnull_islocal");
+	i_notnull_validated = PQfnumber(res, "notnull_validated");
 	i_attoptions = PQfnumber(res, "attoptions");
 	i_attcollation = PQfnumber(res, "attcollation");
 	i_attcompression = PQfnumber(res, "attcompression");
@@ -9094,6 +9099,7 @@ getTableAttrs(Archive *fout, TableInfo *tblinfo, int numTables)
 	 * r is handled by the inner loop.
 	 */
 	curtblindx = -1;
+	appendPQExpBufferChar(invalidnotnulloids, '{');
 	for (int r = 0; r < ntups;)
 	{
 		Oid			attrelid = atooid(PQgetvalue(res, r, i_attrelid));
@@ -9145,6 +9151,7 @@ getTableAttrs(Archive *fout, TableInfo *tblinfo, int numTables)
 		tbinfo->notnull_constrs = (char **) pg_malloc(numatts * sizeof(char *));
 		tbinfo->notnull_noinh = (bool *) pg_malloc(numatts * sizeof(bool));
 		tbinfo->notnull_islocal = (bool *) pg_malloc(numatts * sizeof(bool));
+		tbinfo->notnull_validated = (bool *) pg_malloc(numatts * sizeof(bool));
 		tbinfo->attrdefs = (AttrDefInfo **) pg_malloc(numatts * sizeof(AttrDefInfo *));
 		hasdefaults = false;
 
@@ -9168,12 +9175,33 @@ 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_validated[j] = (PQgetvalue(res, r, i_notnull_validated)[0] == 't');
 
-			/* Handle not-null constraint name and flags */
-			determineNotNullFlags(fout, res, r,
-								  tbinfo, j,
-								  i_notnull_name, i_notnull_noinherit,
-								  i_notnull_islocal);
+			if (tbinfo->notnull_validated[j])
+			{
+				/* Handle not-null constraint name and flags */
+				determineNotNullFlags(fout, res, r,
+									tbinfo, j,
+									i_notnull_name, i_notnull_noinherit,
+									i_notnull_islocal);
+			}
+			else
+			{
+				tbinfo->notnull_constrs[j] = NULL;
+
+				/*
+				 * if column notnull_validated is false, it can also mean that
+				 * column don't have not-null constraint at all. if column have
+				 * NOT NULL NOT VALID then Add the entry into invalidnotnulloids
+				 * list so it can dumped as seperate constraint.
+				*/
+				if (!PQgetisnull(res, r, i_notnull_name))
+				{
+					if (invalidnotnulloids->len > 1)
+						appendPQExpBufferChar(invalidnotnulloids, ',');
+					appendPQExpBuffer(invalidnotnulloids, "%u", tbinfo->dobj.catId.oid);
+				}
+			}
 
 			tbinfo->attoptions[j] = pg_strdup(PQgetvalue(res, r, i_attoptions));
 			tbinfo->attcollation[j] = atooid(PQgetvalue(res, r, i_attcollation));
@@ -9193,6 +9221,7 @@ getTableAttrs(Archive *fout, TableInfo *tblinfo, int numTables)
 			appendPQExpBuffer(tbloids, "%u", tbinfo->dobj.catId.oid);
 		}
 	}
+	appendPQExpBufferChar(invalidnotnulloids, '}');
 
 	PQclear(res);
 
@@ -9326,6 +9355,110 @@ getTableAttrs(Archive *fout, TableInfo *tblinfo, int numTables)
 		PQclear(res);
 	}
 
+	/*
+	 * Get info about table 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->len > 2)
+	{
+		ConstraintInfo *constrs;
+		int         numConstrs;
+		int         i_tableoid;
+		int         i_oid;
+		int         i_conrelid;
+		int         i_conname;
+		int         i_consrc;
+		int         i_conislocal;
+		int         i_convalidated;
+
+		pg_log_info("finding table 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(tbloid)\n"
+						  "JOIN pg_catalog.pg_constraint c ON (src.tbloid = c.conrelid)\n"
+						  "WHERE contype = 'n' AND NOT convalidated "
+						  "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");
+		i_convalidated = PQfnumber(res, "convalidated");
+
+		/* 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);
+
+			tbinfo->checkexprs = constrs + j;
+
+			for (int c = 0; c < numcons; c++, j++)
+			{
+				bool        validated = PQgetvalue(res, j, i_convalidated)[0] == 't';
+
+				Assert(!validated);
+				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');
+
+				/*
+		         * NOT NULL NOT VALID must dumped separately as an ALTER TABLE
+				 * ADD CONSTRAINT entry. because CREATE TABLE can not create
+				 * invalid not null constraint.
+				 */
+				constrs[j].separate = !validated;
+
+				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.
@@ -16620,6 +16753,8 @@ dumpTableSchema(Archive *fout, const TableInfo *tbinfo)
 
 					if (print_notnull)
 					{
+						Assert(tbinfo->notnull_validated[j]);
+
 						if (tbinfo->notnull_constrs[j][0] == '\0')
 							appendPQExpBufferStr(q, " NOT NULL");
 						else
@@ -17939,9 +18074,9 @@ 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 and INVALID NOT NULL constraint on a table */
 
 		/* Ignore if not to be dumped separately, or if it was inherited */
 		if (coninfo->separate && coninfo->conislocal)
@@ -17965,7 +18100,7 @@ dumpConstraint(Archive *fout, const ConstraintInfo *coninfo)
 							 ARCHIVE_OPTS(.tag = tag,
 										  .namespace = tbinfo->dobj.namespace->dobj.name,
 										  .owner = tbinfo->rolname,
-										  .description = "CHECK CONSTRAINT",
+										  .description = coninfo->contype == 'c' ? "CHECK CONSTRAINT" : "CONSTRAINT",
 										  .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 bbdb30b5f54..e0c9bbd64a2 100644
--- a/src/bin/pg_dump/pg_dump.h
+++ b/src/bin/pg_dump/pg_dump.h
@@ -367,6 +367,7 @@ typedef struct _tableInfo
 									 * (pre-v17) */
 	bool	   *notnull_noinh;	/* NOT NULL is NO INHERIT */
 	bool	   *notnull_islocal;	/* true if NOT NULL has local definition */
+	bool	   *notnull_validated;	/* true if NOT NULL is validated */
 	struct _attrDefInfo **attrdefs; /* DEFAULT expressions */
 	struct _constraintInfo *checkexprs; /* CHECK constraints */
 	bool		needs_override; /* has GENERATED ALWAYS AS IDENTITY */
diff --git a/src/bin/psql/describe.c b/src/bin/psql/describe.c
index e6cf468ac9e..09f8f92a59c 100644
--- a/src/bin/psql/describe.c
+++ b/src/bin/psql/describe.c
@@ -3114,7 +3114,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"
@@ -3137,14 +3138,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 ? " NO VALID": "");
 
 				printTableAddFooter(&cont, buf.data);
 			}
diff --git a/src/include/access/tupdesc.h b/src/include/access/tupdesc.h
index 396eeb7a0bb..cc619477b63 100644
--- a/src/include/access/tupdesc.h
+++ b/src/include/access/tupdesc.h
@@ -43,6 +43,7 @@ typedef struct TupleConstr
 	uint16		num_defval;
 	uint16		num_check;
 	bool		has_not_null;
+	bool        has_invalid_not_null;
 	bool		has_generated_stored;
 	bool		has_generated_virtual;
 } TupleConstr;
diff --git a/src/include/catalog/pg_attribute.h b/src/include/catalog/pg_attribute.h
index deaa515fe53..9998b4abdef 100644
--- a/src/include/catalog/pg_attribute.h
+++ b/src/include/catalog/pg_attribute.h
@@ -120,6 +120,9 @@ CATALOG(pg_attribute,1249,AttributeRelationId) BKI_BOOTSTRAP BKI_ROWTYPE_OID(75,
 	/* This flag represents the "NOT NULL" constraint */
 	bool		attnotnull;
 
+	/* This flag represents the "NOT NULL NOT VALID" constraint */
+	bool		attinvalidnotnull BKI_DEFAULT(f);
+
 	/* 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..fe27dde16fa 100644
--- a/src/include/catalog/pg_constraint.h
+++ b/src/include/catalog/pg_constraint.h
@@ -259,14 +259,14 @@ extern char *ChooseConstraintName(const char *name1, const char *name2,
 								  const char *label, Oid namespaceid,
 								  List *others);
 
-extern HeapTuple findNotNullConstraintAttnum(Oid relid, AttrNumber attnum);
-extern HeapTuple findNotNullConstraint(Oid relid, const char *colname);
+extern HeapTuple findNotNullConstraintAttnum(Oid relid, AttrNumber attnum, bool include_invalid);
+extern HeapTuple findNotNullConstraint(Oid relid, const char *colname, bool include_invalid);
 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);
+										   bool include_noinh, bool include_notvalid);
 
 extern void RemoveConstraintById(Oid conId);
 extern void RenameConstraintById(Oid conId, const char *newname);
diff --git a/src/test/regress/expected/constraints.out b/src/test/regress/expected/constraints.out
index 4f39100fcdf..d57d7e09984 100644
--- a/src/test/regress/expected/constraints.out
+++ b/src/test/regress/expected/constraints.out
@@ -896,6 +896,175 @@ Not-null constraints:
     "foobar" NOT NULL "a"
 
 DROP TABLE notnull_tbl1;
+-------tests for 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 INTEGER, b INTEGER);
+INSERT INTO notnull_tbl1 VALUES (NULL, 1),(NULL, 2), (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 |           |          |         | plain   |              | 
+ b      | integer |           |          |         | plain   |              | 
+Not-null constraints:
+    "nn" NOT NULL "a" NO VALID
+
+INSERT INTO notnull_tbl1 VALUES (NULL, 4); --error, invalid not-null constraint forbiden new null values
+ERROR:  null value in column "a" of relation "notnull_tbl1" violates not-null constraint
+DETAIL:  Failing row contains (null, 4).
+ALTER TABLE notnull_tbl1 ADD CONSTRAINT nn1 NOT NULL a not valid no inherit; --error
+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 can't change not-null status
+ERROR:  cannot change VALID status of NOT NULL constraint "nn" on relation "notnull_tbl1"
+HINT:  You may need to use ALTER TABLE VALIDATE CONSTRAINT to validate constraint "nn"
+ALTER TABLE notnull_tbl1 ALTER a SET NOT NULL; --validate it then error out
+ERROR:  column "a" of relation "notnull_tbl1" contains null values
+UPDATE notnull_tbl1 SET a = 100 WHERE b = 1;
+DELETE FROM notnull_tbl1 WHERE b = 2;
+SELECT * FROM notnull_tbl1 ORDER BY a, b;
+  a  | b 
+-----+---
+ 100 | 1
+ 300 | 3
+(2 rows)
+
+-- cannot add primary key on column marked as NOT VALID NOT NULL
+ALTER TABLE notnull_tbl1 ADD PRIMARY KEY (a);
+ERROR:  primary key require an validated NOT NULL constraint
+DETAIL:  Column "a" of table "notnull_tbl1" is marked as NOT VALID NOT NULL constraint
+HINT:  You may try ALTER TABLE VALIDATE CONSTRAINT to validate constraint "nn"
+-- however child table NOT NULL constraints should be valid.
+CREATE TABLE notnull_tbl1_child(a INTEGER, b INTEGER) 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;
+---now we can add primary key
+ALTER TABLE notnull_tbl1 ADD PRIMARY KEY (a);
+DROP TABLE notnull_tbl1_child;
+DROP TABLE notnull_tbl1;
+-- Test the different not null constraint name for parent and child table
+CREATE TABLE notnull_tbl1 (a int);
+ALTER TABLE notnull_tbl1 ADD CONSTRAINT nn_parent NOT NULL a not valid;
+-- parent have valid not null constraint then child table cannot have invalid one
+CREATE TABLE notnull_chld0 (a int);
+ALTER TABLE notnull_chld0 ADD CONSTRAINT nn_chld0 NOT NULL a;
+ALTER TABLE notnull_tbl1 INHERIT notnull_chld0; --error
+ERROR:  column "a" in child table "notnull_tbl1" must be marked NOT NULL
+CREATE TABLE notnull_chld (a int);
+ALTER TABLE notnull_chld ADD CONSTRAINT nn_child NOT NULL a not valid;
+ALTER TABLE notnull_chld INHERIT notnull_tbl1;
+ALTER TABLE notnull_chld VALIDATE CONSTRAINT nn_child;
+-- parents and child not-null will all be validated.
+ALTER TABLE notnull_tbl1 VALIDATE CONSTRAINT nn_parent;
+EXECUTE get_nnconstraint_info('{notnull_tbl1, notnull_chld}');
+   conrelid   |  conname  | convalidated | coninhcount 
+--------------+-----------+--------------+-------------
+ notnull_tbl1 | nn_parent | t            |           0
+ notnull_chld | nn_child  | t            |           1
+(2 rows)
+
+DROP TABLE notnull_chld0;
+DROP TABLE notnull_chld;
+DROP TABLE notnull_tbl1;
+-- 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:  cannot change VALID status of NOT NULL constraint "nn" on relation "inh_parent"
+HINT:  You may need to use ALTER TABLE VALIDATE CONSTRAINT to validate constraint "nn"
+ALTER TABLE inh_child ADD CONSTRAINT nn1 NOT NULL i; --error
+ERROR:  cannot change VALID status of NOT NULL constraint "nn" on relation "inh_child"
+HINT:  You may need to use ALTER TABLE VALIDATE CONSTRAINT to validate constraint "nn"
+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 INTEGER, b INTEGER) 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, table already 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 INTEGER, b INTEGER) PARTITION BY LIST (a);
+ALTER TABLE pp_nn ADD CONSTRAINT pp_nn_notnull NOT NULL 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:  column "a" in child table "pp_nn_1" must be marked NOT NULL
+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;
+DEALLOCATE get_nnconstraint_info;
+-- Create table with NOT NULL INVALID constraint, for pg_upgrade.
+CREATE TABLE notnull_tbl1_upg (a INTEGER, b INTEGER);
+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;
+-- 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
diff --git a/src/test/regress/sql/constraints.sql b/src/test/regress/sql/constraints.sql
index 21ce4177de4..976e8e6941c 100644
--- a/src/test/regress/sql/constraints.sql
+++ b/src/test/regress/sql/constraints.sql
@@ -640,6 +640,120 @@ ALTER TABLE notnull_tbl1 ADD CONSTRAINT foobar NOT NULL a;
 \d+ notnull_tbl1
 DROP TABLE notnull_tbl1;
 
+-------tests for 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 INTEGER, b INTEGER);
+INSERT INTO notnull_tbl1 VALUES (NULL, 1),(NULL, 2), (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
+
+INSERT INTO notnull_tbl1 VALUES (NULL, 4); --error, invalid not-null constraint forbiden new null values
+ALTER TABLE notnull_tbl1 ADD CONSTRAINT nn1 NOT NULL a not valid no inherit; --error
+ALTER TABLE notnull_tbl1 ADD CONSTRAINT nn NOT NULL a; --error can't change not-null status
+ALTER TABLE notnull_tbl1 ALTER a SET NOT NULL; --validate it then error out
+
+UPDATE notnull_tbl1 SET a = 100 WHERE b = 1;
+DELETE FROM notnull_tbl1 WHERE b = 2;
+SELECT * FROM notnull_tbl1 ORDER BY a, b;
+
+-- cannot add primary key on column marked as NOT VALID NOT NULL
+ALTER TABLE notnull_tbl1 ADD PRIMARY KEY (a);
+-- however child table NOT NULL constraints should be valid.
+CREATE TABLE notnull_tbl1_child(a INTEGER, b INTEGER) INHERITS(notnull_tbl1);
+EXECUTE get_nnconstraint_info('{notnull_tbl1_child, notnull_tbl1}');
+
+ALTER TABLE notnull_tbl1 VALIDATE CONSTRAINT nn;
+---now we can add primary key
+ALTER TABLE notnull_tbl1 ADD PRIMARY KEY (a);
+DROP TABLE notnull_tbl1_child;
+DROP TABLE notnull_tbl1;
+
+
+-- Test the different not null constraint name for parent and child table
+CREATE TABLE notnull_tbl1 (a int);
+ALTER TABLE notnull_tbl1 ADD CONSTRAINT nn_parent NOT NULL a not valid;
+
+-- parent have valid not null constraint then child table cannot have invalid one
+CREATE TABLE notnull_chld0 (a int);
+ALTER TABLE notnull_chld0 ADD CONSTRAINT nn_chld0 NOT NULL a;
+ALTER TABLE notnull_tbl1 INHERIT notnull_chld0; --error
+
+CREATE TABLE notnull_chld (a int);
+ALTER TABLE notnull_chld ADD CONSTRAINT nn_child NOT NULL a not valid;
+ALTER TABLE notnull_chld INHERIT notnull_tbl1;
+
+ALTER TABLE notnull_chld VALIDATE CONSTRAINT nn_child;
+-- parents and child not-null will all be validated.
+ALTER TABLE notnull_tbl1 VALIDATE CONSTRAINT nn_parent;
+EXECUTE get_nnconstraint_info('{notnull_tbl1, notnull_chld}');
+DROP TABLE notnull_chld0;
+DROP TABLE notnull_chld;
+DROP TABLE notnull_tbl1;
+
+
+-- 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 INTEGER, b INTEGER) 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, table already 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 INTEGER, b INTEGER) PARTITION BY LIST (a);
+ALTER TABLE pp_nn ADD CONSTRAINT pp_nn_notnull NOT NULL 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;
+
+DEALLOCATE get_nnconstraint_info;
+
+-- Create table with NOT NULL INVALID constraint, for pg_upgrade.
+CREATE TABLE notnull_tbl1_upg (a INTEGER, b INTEGER);
+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;
+-- 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
-- 
2.34.1