0002-Support-NOT-VALID-and-VALIDATE-CONSTRAINT-for-named-.patch

From 8348e55717086f08fb12cefd4fa0156a541867ab Mon Sep 17 00:00:00 2001
From: Rushabh Lathia <rushabh.lathia@enterprisedb.com>
Date: Mon, 10 Feb 2025 16:22:30 +0530
Subject: [PATCH 2/3] Support NOT VALID and VALIDATE CONSTRAINT for named NOT
 NULL constraints.

Commit also add support for pg_dump to dump NOT VALID named NOT NULL
constraints.
---
 src/backend/bootstrap/bootstrap.c         |   4 +-
 src/backend/catalog/heap.c                |   2 +-
 src/backend/catalog/pg_constraint.c       |  31 ++-
 src/backend/commands/tablecmds.c          | 250 ++++++++++++++++++----
 src/backend/executor/execMain.c           |   3 +-
 src/backend/parser/gram.y                 |   4 +-
 src/bin/pg_dump/pg_dump.c                 | 169 +++++++++++++--
 src/bin/pg_dump/pg_dump.h                 |   1 +
 src/bin/pg_dump/t/002_pg_dump.pl          |  17 ++
 src/bin/psql/describe.c                   |  10 +-
 src/include/catalog/pg_attribute.h        |   1 +
 src/include/catalog/pg_constraint.h       |   3 +-
 src/test/regress/expected/constraints.out | 153 +++++++++++++
 src/test/regress/sql/constraints.sql      |  85 ++++++++
 14 files changed, 661 insertions(+), 72 deletions(-)

diff --git a/src/backend/bootstrap/bootstrap.c b/src/backend/bootstrap/bootstrap.c
index 1e95dc32f4..919972dc40 100644
--- a/src/backend/bootstrap/bootstrap.c
+++ b/src/backend/bootstrap/bootstrap.c
@@ -582,10 +582,8 @@ DefineAttr(char *name, char *type, int attnum, int nullness)
 
 	attrtypes[attnum]->atttypmod = -1;
 	attrtypes[attnum]->attislocal = true;
-	/* set default to false */
 	attrtypes[attnum]->attnotnull = ATTRIBUTE_NOTNULL_FALSE;
 
-
 	if (nullness == BOOTCOL_NULL_FORCE_NOT_NULL)
 	{
 		attrtypes[attnum]->attnotnull = ATTRIBUTE_NOTNULL_TRUE;
@@ -699,7 +697,7 @@ InsertOneNull(int i)
 {
 	elog(DEBUG4, "inserting column %d NULL", i);
 	Assert(i >= 0 && i < MAXATTR);
-	if (TupleDescAttr(boot_reldesc->rd_att, i)->attnotnull == ATTRIBUTE_NOTNULL_TRUE)
+	if (TupleDescAttr(boot_reldesc->rd_att, i)->attnotnull != ATTRIBUTE_NOTNULL_FALSE)
 		elog(ERROR,
 			 "NULL value specified for not-null column \"%s\" of relation \"%s\"",
 			 NameStr(TupleDescAttr(boot_reldesc->rd_att, i)->attname),
diff --git a/src/backend/catalog/heap.c b/src/backend/catalog/heap.c
index 6dc94e3693..4337cf3e7e 100644
--- a/src/backend/catalog/heap.c
+++ b/src/backend/catalog/heap.c
@@ -2575,7 +2575,7 @@ AddRelationNewConstraints(Relation rel,
 			 * to add another one; just adjust inheritance status as needed.
 			 */
 			if (AdjustNotNullInheritance(RelationGetRelid(rel), colnum,
-										 is_local, cdef->is_no_inherit))
+										 cdef->conname, is_local, cdef->is_no_inherit))
 				continue;
 
 			/*
diff --git a/src/backend/catalog/pg_constraint.c b/src/backend/catalog/pg_constraint.c
index ac80652baf..e14041b34c 100644
--- a/src/backend/catalog/pg_constraint.c
+++ b/src/backend/catalog/pg_constraint.c
@@ -574,7 +574,7 @@ 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.
+ * 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
@@ -604,13 +604,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)
@@ -628,9 +626,9 @@ 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 the pg_constraint tuple that implements a
+ * 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)
@@ -726,10 +724,13 @@ extractNotNullColumn(HeapTuple constrTup)
  * 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.
+ *
+ * Function add a check for existing INVALID not-null constraint on
+ * the table column.
  */
 bool
 AdjustNotNullInheritance(Oid relid, AttrNumber attnum,
-						 bool is_local, bool is_no_inherit)
+						 const char *conname, bool is_local, bool is_no_inherit)
 {
 	HeapTuple	tup;
 
@@ -753,6 +754,18 @@ 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 an invalid NOT NULL constraint exists on the
+		 * table column and an attempt is made to add another valid NOT NULL
+		 * constraint.
+		 */
+		if (is_local && !conform->convalidated && conname)
+			ereport(ERROR,
+					errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+					errmsg("Invalid NOT NULL constraint \"%s\" exist on relation \"%s\"",
+						   NameStr(conform->conname), get_rel_name(relid)),
+					errhint("Do VALIDATE CONSTRAINT"));
+
 		if (!is_local)
 		{
 			if (pg_add_s16_overflow(conform->coninhcount, 1,
diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c
index c47ac4d1d3..7fff892aba 100644
--- a/src/backend/commands/tablecmds.c
+++ b/src/backend/commands/tablecmds.c
@@ -405,6 +405,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,
@@ -468,6 +471,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,
+						   char newvalue, bool queue_validation,
 						   LOCKMODE lockmode);
 static ObjectAddress ATExecSetNotNull(List **wqueue, Relation rel,
 									  char *constrname, char *colName,
@@ -697,6 +701,7 @@ static List *GetParentedForeignKeyRefs(Relation partition);
 static void ATDetachCheckNoForeignKeyRefs(Relation partition);
 static char GetAttributeCompression(Oid atttypid, const char *compression);
 static char GetAttributeStorage(Oid atttypid, const char *storagemode);
+static bool check_for_invalid_notnull(Oid relid, const char *attname);
 
 
 /* ----------------------------------------------------------------
@@ -1314,7 +1319,8 @@ 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, ATTRIBUTE_NOTNULL_TRUE,
+					   false, NoLock);
 
 	ObjectAddressSet(address, RelationRelationId, relationId);
 
@@ -6190,7 +6196,8 @@ ATRewriteTable(AlteredTableInfo *tab, Oid OIDNewHeap)
 		{
 			Form_pg_attribute attr = TupleDescAttr(newTupDesc, i);
 
-			if (attr->attnotnull == ATTRIBUTE_NOTNULL_TRUE &&
+			if ((attr->attnotnull == ATTRIBUTE_NOTNULL_TRUE ||
+				 attr->attnotnull == ATTRIBUTE_NOTNULL_INVALID) &&
 				!attr->attisdropped)
 				notnull_attrs = lappend_int(notnull_attrs, i);
 		}
@@ -7711,10 +7718,13 @@ ATExecDropNotNull(Relation rel, const char *colName, bool recurse,
  */
 static void
 set_attnotnull(List **wqueue, Relation rel, AttrNumber attnum,
-			   LOCKMODE lockmode)
+			   char newvalue, bool queue_validation, LOCKMODE lockmode)
 {
 	Form_pg_attribute attr;
 
+
+	Assert(!queue_validation || wqueue != NULL);
+
 	CheckAlterTableIsSafe(rel);
 
 	/*
@@ -7725,7 +7735,7 @@ set_attnotnull(List **wqueue, Relation rel, AttrNumber attnum,
 	if (attr->attisdropped)
 		return;
 
-	if (attr->attnotnull == ATTRIBUTE_NOTNULL_FALSE)
+	if (attr->attnotnull != newvalue)
 	{
 		Relation	attr_rel;
 		HeapTuple	tuple;
@@ -7738,15 +7748,17 @@ set_attnotnull(List **wqueue, Relation rel, AttrNumber attnum,
 				 attnum, RelationGetRelid(rel));
 
 		attr = (Form_pg_attribute) GETSTRUCT(tuple);
-		Assert(attr->attnotnull == ATTRIBUTE_NOTNULL_FALSE);
-		attr->attnotnull = ATTRIBUTE_NOTNULL_TRUE;
+		attr->attnotnull = newvalue;
+
 		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.
+		 * Queue later validation of this constraint, if necessary and
+		 * requested by caller.
 		 */
-		if (wqueue && !NotNullImpliedByRelConstraints(rel, attr))
+		if (queue_validation &&
+			newvalue == ATTRIBUTE_NOTNULL_TRUE &&
+			!NotNullImpliedByRelConstraints(rel, attr))
 		{
 			AlteredTableInfo *tab;
 
@@ -7852,6 +7864,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, conForm->conname.data,
+											recurse, recursing, lockmode);
+		}
 
 		if (changed)
 		{
@@ -7914,8 +7935,12 @@ 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 request validation */
+	set_attnotnull(wqueue, rel, attnum,
+				   constraint->skip_validation ?
+				   ATTRIBUTE_NOTNULL_INVALID :
+				   ATTRIBUTE_NOTNULL_TRUE,
+				   !constraint->skip_validation, lockmode);
 
 	/*
 	 * Recurse to propagate the constraint to children that don't have one.
@@ -9357,12 +9382,19 @@ ATPrepAddPrimaryKey(List **wqueue, Relation rel, AlterTableCmd *cmd,
 	}
 
 	/* Insert not-null constraints in the queue for the PK columns */
-	foreach(lc, pkconstr->keys)
+	foreach_node(String, colname, pkconstr->keys)
 	{
 		AlterTableCmd *newcmd;
 		Constraint *nnconstr;
 
-		nnconstr = makeNotNullConstraint(lfirst(lc));
+		/* Verify that the not-null constraint has been validated */
+		if (check_for_invalid_notnull(RelationGetRelid(rel), strVal(colname)))
+			ereport(ERROR,
+					errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+					errmsg("column \"%s\" of table \"%s\" is marked as NOT VALID NOT NULL constraint",
+						   strVal(colname), RelationGetRelationName(rel)));
+
+		nnconstr = makeNotNullConstraint(colname);
 
 		newcmd = makeNode(AlterTableCmd);
 		newcmd->subtype = AT_AddConstraint;
@@ -9373,6 +9405,30 @@ ATPrepAddPrimaryKey(List **wqueue, Relation rel, AlterTableCmd *cmd,
 	}
 }
 
+/*
+ * This function checks for any invalid NOT NULL constraint on the given
+ * relation and attribute name. It returns true if found, false otherwise.
+ */
+static bool
+check_for_invalid_notnull(Oid relid, const char *attname)
+{
+	HeapTuple	tuple;
+	bool		retval = false;
+
+	tuple = SearchSysCache2(ATTNAME,
+							ObjectIdGetDatum(relid),
+							CStringGetDatum(attname));
+	if (!HeapTupleIsValid(tuple))
+		return false;
+	if (!((Form_pg_attribute) GETSTRUCT(tuple))->attisdropped &&
+		((Form_pg_attribute) GETSTRUCT(tuple))->attnotnull == ATTRIBUTE_NOTNULL_INVALID)
+		retval = true;
+
+	ReleaseSysCache(tuple);
+
+	return retval;
+}
+
 /*
  * ALTER TABLE ADD INDEX
  *
@@ -9740,7 +9796,12 @@ ATAddCheckNNConstraint(List **wqueue, AlteredTableInfo *tab, Relation rel,
 		 * phase 3 to verify existing rows, if needed.
 		 */
 		if (constr->contype == CONSTR_NOTNULL)
-			set_attnotnull(wqueue, rel, ccon->attnum, lockmode);
+			set_attnotnull(wqueue, rel, ccon->attnum,
+						   ccon->skip_validation ?
+						   ATTRIBUTE_NOTNULL_INVALID :
+						   ATTRIBUTE_NOTNULL_TRUE,
+						   !ccon->skip_validation,
+						   lockmode);
 
 		ObjectAddressSet(address, ConstraintRelationId, ccon->conoid);
 	}
@@ -12154,10 +12215,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, not-null, or check constraint",
 						constrName, RelationGetRelationName(rel))));
 
 	if (!con->conenforced)
@@ -12176,6 +12238,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);
 	}
@@ -12305,9 +12372,7 @@ QueueCheckConstraintValidation(List **wqueue, Relation conrel, Relation rel,
 	AlteredTableInfo *tab;
 	HeapTuple	copyTuple;
 	Form_pg_constraint copy_con;
-
 	List	   *children = NIL;
-	ListCell   *child;
 	NewConstraint *newcon;
 	Datum		val;
 	char	   *conbin;
@@ -12316,24 +12381,19 @@ QueueCheckConstraintValidation(List **wqueue, Relation conrel, Relation rel,
 	Assert(con->contype == CONSTRAINT_CHECK);
 
 	/*
-	 * If we're recursing, the parent has already done this, so skip it. Also,
-	 * if the constraint is a NO INHERIT constraint, we shouldn't try to look
-	 * for it in the children.
-	 */
-	if (!recursing && !con->connoinherit)
-		children = find_all_inheritors(RelationGetRelid(rel),
-									   lockmode, NULL);
-
-	/*
-	 * For CHECK constraints, we must ensure that we only mark the constraint
-	 * as validated on the parent if it's already validated on the children.
+	 * 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.
 	 */
-	foreach(child, children)
+	if (!recursing && !con->connoinherit)
+		children = find_all_inheritors(RelationGetRelid(rel), lockmode, NULL);
+	foreach_oid(childoid, children)
 	{
-		Oid			childoid = lfirst_oid(child);
 		Relation	childrel;
 
 		if (childoid == RelationGetRelid(rel))
@@ -12392,6 +12452,112 @@ 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);
+		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;
+
+	/*
+	 * 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);
+
+	/* Also flip attnotnull */
+	set_attnotnull(wqueue, rel, attnum, ATTRIBUTE_NOTNULL_TRUE, false,
+				   lockmode);
+
+	InvokeObjectPostAlterHook(ConstraintRelationId, con->oid, 0);
+
+	heap_freetuple(copyTuple);
+}
+
 /*
  * transformColumnNameList - transform list of column names
  *
@@ -13263,7 +13429,7 @@ dropconstraint_internal(Relation rel, HeapTuple constraintTup, DropBehavior beha
 						   RelationGetRelationName(rel)));
 
 		/* All good -- reset attnotnull if needed */
-		if (attForm->attnotnull == ATTRIBUTE_NOTNULL_TRUE)
+		if (attForm->attnotnull != ATTRIBUTE_NOTNULL_FALSE)
 		{
 			attForm->attnotnull = ATTRIBUTE_NOTNULL_FALSE;
 			CatalogTupleUpdate(attrel, &atttup->t_self, atttup);
@@ -16599,12 +16765,20 @@ MergeAttributesIntoExisting(Relation child_rel, Relation parent_rel, bool ispart
 
 				contup = findNotNullConstraintAttnum(RelationGetRelid(parent_rel),
 													 parent_att->attnum);
-				if (HeapTupleIsValid(contup) &&
-					!((Form_pg_constraint) GETSTRUCT(contup))->connoinherit)
-					ereport(ERROR,
-							errcode(ERRCODE_DATATYPE_MISMATCH),
-							errmsg("column \"%s\" in child table \"%s\" must be marked NOT NULL",
-								   parent_attname, RelationGetRelationName(child_rel)));
+				if (HeapTupleIsValid(contup))
+				{
+					Form_pg_constraint	childcon;
+
+					childcon = (Form_pg_constraint) GETSTRUCT(contup);
+					if (!childcon->connoinherit)
+						ereport(ERROR,
+								errcode(ERRCODE_DATATYPE_MISMATCH),
+								errmsg("column \"%s\" in child table \"%s\" must be marked NOT NULL",
+									   parent_attname, RelationGetRelationName(child_rel)));
+
+					if (!childcon->convalidated)
+						elog(WARNING, "found an invalid constraint");
+				}
 			}
 
 			/*
diff --git a/src/backend/executor/execMain.c b/src/backend/executor/execMain.c
index d582231ab6..4b0446b00d 100644
--- a/src/backend/executor/execMain.c
+++ b/src/backend/executor/execMain.c
@@ -2070,7 +2070,8 @@ ExecConstraints(ResultRelInfo *resultRelInfo,
 		{
 			Form_pg_attribute att = TupleDescAttr(tupdesc, attrChk - 1);
 
-			if (att->attnotnull == ATTRIBUTE_NOTNULL_TRUE &&
+			if ((att->attnotnull == ATTRIBUTE_NOTNULL_TRUE ||
+				 att->attnotnull == ATTRIBUTE_NOTNULL_INVALID) &&
 				slot_attisnull(slot, attrChk))
 			{
 				char	   *val_desc;
diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y
index 7d99c9355c..b3fd4183b2 100644
--- a/src/backend/parser/gram.y
+++ b/src/backend/parser/gram.y
@@ -4196,9 +4196,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/bin/pg_dump/pg_dump.c b/src/bin/pg_dump/pg_dump.c
index afd7928717..7f0c39db88 100644
--- a/src/bin/pg_dump/pg_dump.c
+++ b/src/bin/pg_dump/pg_dump.c
@@ -8830,6 +8830,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;
@@ -8846,6 +8847,7 @@ getTableAttrs(Archive *fout, TableInfo *tblinfo, int numTables)
 	int			i_attlen;
 	int			i_attalign;
 	int			i_attislocal;
+	int			i_notnull_valid;
 	int			i_notnull_name;
 	int			i_notnull_noinherit;
 	int			i_notnull_islocal;
@@ -8942,11 +8944,13 @@ getTableAttrs(Archive *fout, TableInfo *tblinfo, int numTables)
 	 */
 	if (fout->remoteVersion >= 180000)
 		appendPQExpBufferStr(q,
+							 "CASE WHEN a.attnotnull = 'i' THEN 'f' ELSE 't' END AS notnull_valid,\n"
 							 "co.conname AS notnull_name,\n"
 							 "co.connoinherit AS notnull_noinherit,\n"
 							 "co.conislocal AS notnull_islocal,\n");
 	else
 		appendPQExpBufferStr(q,
+							 "'t' AS notnull_valid,\n"
 							 "CASE WHEN a.attnotnull THEN '' ELSE NULL END AS notnull_name,\n"
 							 "false AS notnull_noinherit,\n"
 							 "a.attislocal AS notnull_islocal,\n");
@@ -9021,6 +9025,7 @@ getTableAttrs(Archive *fout, TableInfo *tblinfo, int numTables)
 	i_attlen = PQfnumber(res, "attlen");
 	i_attalign = PQfnumber(res, "attalign");
 	i_attislocal = PQfnumber(res, "attislocal");
+	i_notnull_valid = PQfnumber(res, "notnull_valid");
 	i_notnull_name = PQfnumber(res, "notnull_name");
 	i_notnull_noinherit = PQfnumber(res, "notnull_noinherit");
 	i_notnull_islocal = PQfnumber(res, "notnull_islocal");
@@ -9040,6 +9045,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));
@@ -9088,6 +9094,7 @@ getTableAttrs(Archive *fout, TableInfo *tblinfo, int numTables)
 		tbinfo->attcompression = (char *) pg_malloc(numatts * sizeof(char));
 		tbinfo->attfdwoptions = (char **) pg_malloc(numatts * sizeof(char *));
 		tbinfo->attmissingval = (char **) pg_malloc(numatts * sizeof(char *));
+		tbinfo->notnull_valid = (bool *) pg_malloc(numatts * sizeof(bool));
 		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));
@@ -9114,12 +9121,29 @@ 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_valid[j] = (PQgetvalue(res, r, i_notnull_valid)[0] == 't');
 
-			/* Handle not-null constraint name and flags */
-			determineNotNullFlags(fout, res, r,
-								  tbinfo, j,
-								  i_notnull_name, i_notnull_noinherit,
-								  i_notnull_islocal);
+			/*
+			 * Dump the invalid NOT NULL constraint like the Check constraints
+			 */
+			if (tbinfo->notnull_valid[j])
+				/* Handle not-null constraint name and flags */
+				determineNotNullFlags(fout, res, r,
+									  tbinfo, j,
+									  i_notnull_name, i_notnull_noinherit,
+									  i_notnull_islocal);
+			else if (!PQgetisnull(res, r, i_notnull_name))
+			{
+				/*
+				 * Add the entry into invalidnotnull list so NOT NULL
+				 * constraint get dump as separate constraints.
+				 */
+				if (invalidnotnulloids->len > 1)	/* do we have more than
+													 * the '{'? */
+					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));
@@ -9141,6 +9165,7 @@ getTableAttrs(Archive *fout, TableInfo *tblinfo, int numTables)
 	}
 
 	PQclear(res);
+	appendPQExpBufferChar(invalidnotnulloids, '}');
 
 	/*
 	 * Now get info about column defaults.  This is skipped for a data-only
@@ -9404,6 +9429,128 @@ getTableAttrs(Archive *fout, TableInfo *tblinfo, int numTables)
 		PQclear(res);
 	}
 
+	/*
+	 * Get info about table INVALID NOT NULL 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 convalidated = 'f'"
+						  "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';
+
+				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');
+
+				/*
+				 * An unvalidated constraint needs to be dumped separately, so
+				 * that potentially-violating existing data is loaded before
+				 * the constraint.
+				 */
+				constrs[j].separate = !validated;
+
+				constrs[j].dobj.dump = tbinfo->dobj.dump;
+
+				/*
+				 * Mark the constraint as needing to appear before the table
+				 * --- this is so that any other dependencies of the
+				 * constraint will be emitted before we try to create the
+				 * table.  If the constraint is to be dumped separately, it
+				 * will be dumped after data is loaded anyway, so don't do it.
+				 * (There's an automatic dependency in the opposite direction
+				 * anyway, so don't need to add one manually here.)
+				 */
+				if (!constrs[j].separate)
+					addObjectDependency(&tbinfo->dobj,
+										constrs[j].dobj.dumpId);
+
+				/*
+				 * We will detect later whether the constraint must be split
+				 * out from the table definition.
+				 */
+			}
+		}
+		PQclear(res);
+	}
+
 	destroyPQExpBuffer(q);
 	destroyPQExpBuffer(tbloids);
 	destroyPQExpBuffer(checkoids);
@@ -16515,7 +16662,8 @@ dumpTableSchema(Archive *fout, const TableInfo *tbinfo)
 					 * defined, or if binary upgrade.  (In the latter case, we
 					 * reset conislocal below.)
 					 */
-					print_notnull = (tbinfo->notnull_constrs[j] != NULL &&
+					print_notnull = (tbinfo->notnull_valid[j] &&
+									 tbinfo->notnull_constrs[j] != NULL &&
 									 (tbinfo->notnull_islocal[j] ||
 									  dopt->binary_upgrade ||
 									  tbinfo->ispartition));
@@ -16577,11 +16725,6 @@ dumpTableSchema(Archive *fout, const TableInfo *tbinfo)
 											  tbinfo->attrdefs[j]->adef_expr);
 					}
 
-					print_notnull = (tbinfo->notnull_constrs[j] != NULL &&
-									 (tbinfo->notnull_islocal[j] ||
-									  dopt->binary_upgrade ||
-									  tbinfo->ispartition));
-
 					if (print_notnull)
 					{
 						if (tbinfo->notnull_constrs[j][0] == '\0')
@@ -17903,9 +18046,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/INVALID_NOTNULL constraint on a table */
 
 		/* Ignore if not to be dumped separately, or if it was inherited */
 		if (coninfo->separate && coninfo->conislocal)
diff --git a/src/bin/pg_dump/pg_dump.h b/src/bin/pg_dump/pg_dump.h
index f08f5905aa..a393d99333 100644
--- a/src/bin/pg_dump/pg_dump.h
+++ b/src/bin/pg_dump/pg_dump.h
@@ -361,6 +361,7 @@ typedef struct _tableInfo
 	char	   *attcompression; /* per-attribute compression method */
 	char	  **attfdwoptions;	/* per-attribute fdw options */
 	char	  **attmissingval;	/* per attribute missing value */
+	bool	   *notnull_valid;	/* NOT NULL status */
 	char	  **notnull_constrs;	/* NOT NULL constraint names. If null,
 									 * there isn't one on this column. If
 									 * empty string, unnamed constraint
diff --git a/src/bin/pg_dump/t/002_pg_dump.pl b/src/bin/pg_dump/t/002_pg_dump.pl
index 3945e4f0e2..ce88865532 100644
--- a/src/bin/pg_dump/t/002_pg_dump.pl
+++ b/src/bin/pg_dump/t/002_pg_dump.pl
@@ -1099,6 +1099,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 e6cf468ac9..f48b24ed38 100644
--- a/src/bin/psql/describe.c
+++ b/src/bin/psql/describe.c
@@ -2107,7 +2107,7 @@ describeOneTableDetails(const char *schemaname,
 			printTableAddCell(&cont, PQgetvalue(res, i, attcoll_col), false, false);
 
 			printTableAddCell(&cont,
-							  strcmp(PQgetvalue(res, i, attnotnull_col), "t") == 0 ? "not null" : "",
+							  strcmp(PQgetvalue(res, i, attnotnull_col), "f") == 0 ? "" : "not null",
 							  false, false);
 
 			identity = PQgetvalue(res, i, attidentity_col);
@@ -3114,7 +3114,7 @@ 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, 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"
@@ -3138,13 +3138,15 @@ describeOneTableDetails(const char *schemaname,
 				bool		islocal = PQgetvalue(result, i, 3)[0] == 't';
 				bool		inherited = PQgetvalue(result, i, 4)[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)") : "",
+								  PQgetvalue(result, i, 5)[0] == 'f' ?
+								  " NOT VALID " : "");
 
 				printTableAddFooter(&cont, buf.data);
 			}
diff --git a/src/include/catalog/pg_attribute.h b/src/include/catalog/pg_attribute.h
index b51a267095..a01a0fa96d 100644
--- a/src/include/catalog/pg_attribute.h
+++ b/src/include/catalog/pg_attribute.h
@@ -229,6 +229,7 @@ MAKE_SYSCACHE(ATTNUM, pg_attribute_relid_attnum_index, 128);
 
 #define		  ATTRIBUTE_NOTNULL_TRUE		't'
 #define		  ATTRIBUTE_NOTNULL_FALSE		'f'
+#define		  ATTRIBUTE_NOTNULL_INVALID		'i'
 
 #endif							/* EXPOSE_TO_CLIENT_CODE */
 
diff --git a/src/include/catalog/pg_constraint.h b/src/include/catalog/pg_constraint.h
index 6da164e7e4..0e01ff1bbb 100644
--- a/src/include/catalog/pg_constraint.h
+++ b/src/include/catalog/pg_constraint.h
@@ -264,7 +264,8 @@ 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);
+									 const char *conname, bool is_local,
+									 bool is_no_inherit);
 extern List *RelationGetNotNullConstraints(Oid relid, bool cooked,
 										   bool include_noinh);
 
diff --git a/src/test/regress/expected/constraints.out b/src/test/regress/expected/constraints.out
index 692a69fe45..e5f2d78686 100644
--- a/src/test/regress/expected/constraints.out
+++ b/src/test/regress/expected/constraints.out
@@ -897,6 +897,159 @@ Not-null constraints:
     "foobar" NOT NULL "a"
 
 DROP TABLE notnull_tbl1;
+-- verify NOT NULL VALID/NOT VALID
+CREATE TABLE notnull_tbl1 (a INTEGER, b INTEGER);
+INSERT INTO notnull_tbl1 VALUES (NULL, 1);
+INSERT INTO notnull_tbl1 VALUES (NULL, 2);
+INSERT INTO notnull_tbl1 VALUES (300, 3);
+-- Below statement should throw an error
+ALTER TABLE notnull_tbl1 ADD CONSTRAINT nn NOT NULL a;
+ERROR:  column "a" of relation "notnull_tbl1" contains null values
+ALTER TABLE notnull_tbl1 ADD CONSTRAINT nn NOT NULL a NOT VALID;
+\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 
+
+-- Try to insert new record with NULL, should throw an error
+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).
+-- SELECT NULL values for COLUMN a, should return 2 records.
+SELECT * FROM notnull_tbl1 WHERE a is NULL;
+ a | b 
+---+---
+   | 1
+   | 2
+(2 rows)
+
+-- UPDATE the one of the NULL values
+UPDATE notnull_tbl1 SET a = 100 WHERE b = 1;
+-- DELETE the record (NULL, 2)
+DELETE FROM notnull_tbl1 WHERE b = 2;
+SELECT * FROM notnull_tbl1;
+  a  | b 
+-----+---
+ 300 | 3
+ 100 | 1
+(2 rows)
+
+-- Try to add primary key on table column marked as NOT VALID NOT NULL
+-- constraint. This should throw an error.
+ALTER TABLE notnull_tbl1 add primary key (a);
+ERROR:  column "a" of table "notnull_tbl1" is marked as NOT VALID NOT NULL constraint
+-- INHERITS table having NOT VALID NOT NULL constraints.
+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
+-- Child table NOT NULL constraints should be valid.
+SELECT conname, convalidated FROM pg_catalog.pg_constraint WHERE conrelid = 'notnull_tbl1_child'::regclass;
+ conname | convalidated 
+---------+--------------
+ nn      | t
+(1 row)
+
+DROP TABLE notnull_tbl1_child;
+ALTER TABLE notnull_tbl1 validate constraint nn;
+\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"
+
+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;
+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;
+-- This statement should validate not null constraint for parent as well as
+-- child.
+ALTER TABLE notnull_tbl1 VALIDATE CONSTRAINT nn_parent;
+SELECT conname, convalidated FROM pg_catalog.pg_constraint WHERE conrelid
+in ('notnull_tbl1'::regclass, 'notnull_chld'::regclass);
+  conname  | convalidated 
+-----------+--------------
+ nn_parent | t
+ nn_child  | t
+(2 rows)
+
+DROP TABLE notnull_tbl1 CASCADE;
+NOTICE:  drop cascades to table notnull_chld
+-- test to throw an error when trying to add another NOT NULL
+-- on the table column with INVALID NOT NULL constraint.
+CREATE TABLE notnull_tbl1 (a INTEGER);
+INSERT INTO notnull_tbl1 VALUES ( NULL );
+ALTER TABLE notnull_tbl1 ADD CONSTRAINT nn1 NOT NULL a not valid;
+-- Should throw an error
+ALTER TABLE notnull_tbl1 ADD CONSTRAINT nn NOT NULL a;
+ERROR:  Invalid NOT NULL constraint "nn1" exist on relation "notnull_tbl1"
+HINT:  Do VALIDATE CONSTRAINT
+-- This should throw an error while validating the NOT NULL
+-- constraint.
+ALTER TABLE notnull_tbl1 ALTER a SET NOT NULL;
+ERROR:  column "a" of relation "notnull_tbl1" contains null values
+-- After deleting null values, SET NOT NULL should work
+DELETE FROM notnull_tbl1;
+ALTER TABLE notnull_tbl1 ALTER a SET NOT NULL;
+SELECT conname, convalidated FROM pg_catalog.pg_constraint WHERE conrelid
+in ('notnull_tbl1'::regclass);
+ conname | convalidated 
+---------+--------------
+ nn1     | t
+(1 row)
+
+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 ALTER i SET NOT NULL;
+SELECT conrelid::regclass, conname, convalidated, coninhcount FROM pg_catalog.pg_constraint WHERE conrelid
+in ('inh_parent'::regclass, 'inh_child'::regclass, 'inh_grandchild'::regclass) ORDER BY 1;
+    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;
+-- 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);
+INSERT INTO notnull_tbl1_upg VALUES (NULL, 2);
+INSERT INTO notnull_tbl1_upg VALUES (300, 3);
+ALTER TABLE notnull_tbl1_upg ADD CONSTRAINT nn NOT NULL a NOT VALID;
+-- Verify NOT NULL VALID/NOT VALID with partition table.
+CREATE TABLE notnull_tbl1 (a INTEGER, b INTEGER) PARTITION BY RANGE (a);
+ALTER TABLE notnull_tbl1 ADD CONSTRAINT notnull_con NOT NULL a NOT VALID;
+CREATE TABLE notnull_tbl1_1 PARTITION OF notnull_tbl1 FOR VALUES FROM (0) TO (10);
+CREATE TABLE notnull_tbl1_2 PARTITION OF notnull_tbl1 FOR VALUES FROM (20) TO (30);
+-- Parent table NOT NULL constraints will be marked as validated false, where
+-- for child table it will be true
+SELECT conrelid::regclass, conname, convalidated FROM pg_catalog.pg_constraint WHERE
+conrelid IN ('notnull_tbl1'::regclass, 'notnull_tbl1_1'::regclass, 'notnull_tbl1_2'::regclass);
+    conrelid    |   conname   | convalidated 
+----------------+-------------+--------------
+ notnull_tbl1   | notnull_con | f
+ notnull_tbl1_1 | notnull_con | t
+ notnull_tbl1_2 | notnull_con | t
+(3 rows)
+
+DROP TABLE notnull_tbl1, notnull_tbl1_1, notnull_tbl1_2;
 -- 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 d6742f83fb..2b95d08684 100644
--- a/src/test/regress/sql/constraints.sql
+++ b/src/test/regress/sql/constraints.sql
@@ -641,6 +641,91 @@ ALTER TABLE notnull_tbl1 ADD CONSTRAINT foobar NOT NULL a;
 \d+ notnull_tbl1
 DROP TABLE notnull_tbl1;
 
+-- verify NOT NULL VALID/NOT VALID
+CREATE TABLE notnull_tbl1 (a INTEGER, b INTEGER);
+INSERT INTO notnull_tbl1 VALUES (NULL, 1);
+INSERT INTO notnull_tbl1 VALUES (NULL, 2);
+INSERT INTO notnull_tbl1 VALUES (300, 3);
+-- Below statement should throw an error
+ALTER TABLE notnull_tbl1 ADD CONSTRAINT nn NOT NULL a;
+ALTER TABLE notnull_tbl1 ADD CONSTRAINT nn NOT NULL a NOT VALID;
+\d+ notnull_tbl1
+-- Try to insert new record with NULL, should throw an error
+INSERT INTO notnull_tbl1 VALUES (NULL, 4);
+-- SELECT NULL values for COLUMN a, should return 2 records.
+SELECT * FROM notnull_tbl1 WHERE a is NULL;
+-- UPDATE the one of the NULL values
+UPDATE notnull_tbl1 SET a = 100 WHERE b = 1;
+-- DELETE the record (NULL, 2)
+DELETE FROM notnull_tbl1 WHERE b = 2;
+SELECT * FROM notnull_tbl1;
+-- Try to add primary key on table column marked as NOT VALID NOT NULL
+-- constraint. This should throw an error.
+ALTER TABLE notnull_tbl1 add primary key (a);
+-- INHERITS table having NOT VALID NOT NULL constraints.
+CREATE TABLE notnull_tbl1_child(a INTEGER, b INTEGER) INHERITS(notnull_tbl1);
+-- Child table NOT NULL constraints should be valid.
+SELECT conname, convalidated FROM pg_catalog.pg_constraint WHERE conrelid = 'notnull_tbl1_child'::regclass;
+DROP TABLE notnull_tbl1_child;
+ALTER TABLE notnull_tbl1 validate constraint nn;
+\d+ notnull_tbl1
+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;
+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;
+-- This statement should validate not null constraint for parent as well as
+-- child.
+ALTER TABLE notnull_tbl1 VALIDATE CONSTRAINT nn_parent;
+SELECT conname, convalidated FROM pg_catalog.pg_constraint WHERE conrelid
+in ('notnull_tbl1'::regclass, 'notnull_chld'::regclass);
+DROP TABLE notnull_tbl1 CASCADE;
+-- test to throw an error when trying to add another NOT NULL
+-- on the table column with INVALID NOT NULL constraint.
+CREATE TABLE notnull_tbl1 (a INTEGER);
+INSERT INTO notnull_tbl1 VALUES ( NULL );
+ALTER TABLE notnull_tbl1 ADD CONSTRAINT nn1 NOT NULL a not valid;
+-- Should throw an error
+ALTER TABLE notnull_tbl1 ADD CONSTRAINT nn NOT NULL a;
+-- This should throw an error while validating the NOT NULL
+-- constraint.
+ALTER TABLE notnull_tbl1 ALTER a SET NOT NULL;
+-- After deleting null values, SET NOT NULL should work
+DELETE FROM notnull_tbl1;
+ALTER TABLE notnull_tbl1 ALTER a SET NOT NULL;
+SELECT conname, convalidated FROM pg_catalog.pg_constraint WHERE conrelid
+in ('notnull_tbl1'::regclass);
+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 ALTER i SET NOT NULL;
+SELECT conrelid::regclass, conname, convalidated, coninhcount FROM pg_catalog.pg_constraint WHERE conrelid
+in ('inh_parent'::regclass, 'inh_child'::regclass, 'inh_grandchild'::regclass) ORDER BY 1;
+drop table inh_parent, inh_child, inh_grandchild;
+-- 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);
+INSERT INTO notnull_tbl1_upg VALUES (NULL, 2);
+INSERT INTO notnull_tbl1_upg VALUES (300, 3);
+ALTER TABLE notnull_tbl1_upg ADD CONSTRAINT nn NOT NULL a NOT VALID;
+
+-- Verify NOT NULL VALID/NOT VALID with partition table.
+CREATE TABLE notnull_tbl1 (a INTEGER, b INTEGER) PARTITION BY RANGE (a);
+ALTER TABLE notnull_tbl1 ADD CONSTRAINT notnull_con NOT NULL a NOT VALID;
+CREATE TABLE notnull_tbl1_1 PARTITION OF notnull_tbl1 FOR VALUES FROM (0) TO (10);
+CREATE TABLE notnull_tbl1_2 PARTITION OF notnull_tbl1 FOR VALUES FROM (20) TO (30);
+-- Parent table NOT NULL constraints will be marked as validated false, where
+-- for child table it will be true
+SELECT conrelid::regclass, conname, convalidated FROM pg_catalog.pg_constraint WHERE
+conrelid IN ('notnull_tbl1'::regclass, 'notnull_tbl1_1'::regclass, 'notnull_tbl1_2'::regclass);
+DROP TABLE notnull_tbl1, notnull_tbl1_1, notnull_tbl1_2;
+
 -- 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.43.0