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

From 02a3fbe6fedadde45374fc9cbb73c32b7dd3fce3 Mon Sep 17 00:00:00 2001
From: Rushabh Lathia <rushabh.lathia@enterprisedb.com>
Date: Thu, 20 Mar 2025 11:22:46 +0530
Subject: [PATCH 1/2] 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/access/common/tupdesc.c       |   6 +
 src/backend/bootstrap/bootstrap.c         |   1 +
 src/backend/catalog/heap.c                |   3 +-
 src/backend/catalog/pg_constraint.c       |  17 +-
 src/backend/commands/tablecmds.c          | 216 +++++++++++++++++++++-
 src/backend/optimizer/util/plancat.c      |   3 +-
 src/backend/parser/gram.y                 |   4 +-
 src/bin/psql/describe.c                   |   7 +-
 src/include/access/tupdesc.h              |   2 +
 src/include/catalog/pg_attribute.h        |   3 +
 src/include/catalog/pg_constraint.h       |   3 +-
 src/test/regress/expected/constraints.out | 153 +++++++++++++++
 src/test/regress/sql/constraints.sql      |  85 +++++++++
 13 files changed, 484 insertions(+), 19 deletions(-)

diff --git a/src/backend/access/common/tupdesc.c b/src/backend/access/common/tupdesc.c
index ed2195f14b2..ed269d26090 100644
--- a/src/backend/access/common/tupdesc.c
+++ b/src/backend/access/common/tupdesc.c
@@ -75,6 +75,7 @@ populate_compact_attribute_internal(Form_pg_attribute src,
 	dst->attisdropped = src->attisdropped;
 	dst->attgenerated = (src->attgenerated != '\0');
 	dst->attnotnull = src->attnotnull;
+	dst->attinvalidnotnull = src->attinvalidnotnull;
 
 	switch (src->attalign)
 	{
@@ -252,6 +253,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 +300,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';
@@ -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';
@@ -841,6 +846,7 @@ TupleDescInitEntry(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..3a258ab4e69 100644
--- a/src/backend/catalog/heap.c
+++ b/src/backend/catalog/heap.c
@@ -753,6 +753,7 @@ InsertPgAttributeTuples(Relation pg_attribute_rel,
 		slot[slotCount]->tts_values[Anum_pg_attribute_attstorage - 1] = CharGetDatum(attrs->attstorage);
 		slot[slotCount]->tts_values[Anum_pg_attribute_attcompression - 1] = CharGetDatum(attrs->attcompression);
 		slot[slotCount]->tts_values[Anum_pg_attribute_attnotnull - 1] = BoolGetDatum(attrs->attnotnull);
+		slot[slotCount]->tts_values[Anum_pg_attribute_attinvalidnotnull - 1] = BoolGetDatum(attrs->attinvalidnotnull);
 		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);
@@ -2626,7 +2627,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 ac80652baf2..a1200ac8abb 100644
--- a/src/backend/catalog/pg_constraint.c
+++ b/src/backend/catalog/pg_constraint.c
@@ -609,8 +609,6 @@ findNotNullConstraintAttnum(Oid relid, AttrNumber attnum)
 		 */
 		if (con->contype != CONSTRAINT_NOTNULL)
 			continue;
-		if (!con->convalidated)
-			continue;
 
 		conkey = extractNotNullColumn(conTup);
 		if (conkey != attnum)
@@ -729,7 +727,8 @@ extractNotNullColumn(HeapTuple constrTup)
  */
 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 +752,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 18ff8956577..5f9d02ebe84 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,
@@ -473,7 +476,9 @@ static void add_column_collation_dependency(Oid relid, int32 attnum, Oid collid)
 static ObjectAddress ATExecDropNotNull(Relation rel, const char *colName, bool recurse,
 									   LOCKMODE lockmode);
 static void set_attnotnull(List **wqueue, Relation rel, AttrNumber attnum,
-						   LOCKMODE lockmode);
+						   bool skip_validation, LOCKMODE lockmode);
+static void set_attinvalidnotnull(Relation rel, AttrNumber attnum,
+								  bool attinvalidnotnull);
 static ObjectAddress ATExecSetNotNull(List **wqueue, Relation rel,
 									  char *constrname, char *colName,
 									  bool recurse, bool recursing,
@@ -710,6 +715,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);
 
 
 /* ----------------------------------------------------------------
@@ -1315,7 +1321,7 @@ DefineRelation(CreateStmt *stmt, char relkind, Oid ownerId,
 	nncols = AddRelationNotNullConstraints(rel, stmt->nnconstraints,
 										   old_notnulls);
 	foreach_int(attrnum, nncols)
-		set_attnotnull(NULL, rel, attrnum, NoLock);
+		set_attnotnull(NULL, rel, attrnum, false, NoLock);
 
 	ObjectAddressSet(address, RelationRelationId, relationId);
 
@@ -7729,6 +7735,45 @@ ATExecDropNotNull(Relation rel, const char *colName, bool recurse,
 	return address;
 }
 
+static void
+set_attinvalidnotnull(Relation rel, AttrNumber attnum,
+					  bool attinvalidnotnull)
+{
+	Form_pg_attribute attr;
+
+	CheckAlterTableIsSafe(rel);
+
+	/*
+	 * Exit quickly by testing attnotnull from the tupledesc's copy of the
+	 * attribute.
+	 */
+	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.
@@ -7739,7 +7784,7 @@ ATExecDropNotNull(Relation rel, const char *colName, bool recurse,
  */
 static void
 set_attnotnull(List **wqueue, Relation rel, AttrNumber attnum,
-			   LOCKMODE lockmode)
+			   bool skip_validation, LOCKMODE lockmode)
 {
 	Form_pg_attribute attr;
 
@@ -7768,20 +7813,23 @@ set_attnotnull(List **wqueue, Relation rel, AttrNumber attnum,
 		attr = (Form_pg_attribute) GETSTRUCT(tuple);
 		Assert(!attr->attnotnull);
 		attr->attnotnull = true;
-		CatalogTupleUpdate(attr_rel, &tuple->t_self, tuple);
 
 		/*
 		 * If the nullness isn't already proven by validated constraints, have
 		 * ALTER TABLE phase 3 test for it.
 		 */
-		if (wqueue && !NotNullImpliedByRelConstraints(rel, attr))
+		if (!skip_validation &&
+			wqueue && !NotNullImpliedByRelConstraints(rel, attr))
 		{
 			AlteredTableInfo *tab;
 
 			tab = ATGetQueueEntry(wqueue, rel);
 			tab->verify_new_notnull = true;
 		}
+		else
+			attr->attinvalidnotnull = true;
 
+		CatalogTupleUpdate(attr_rel, &tuple->t_self, tuple);
 		CommandCounterIncrement();
 
 		table_close(attr_rel, RowExclusiveLock);
@@ -7880,6 +7928,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)
 		{
@@ -7943,7 +8000,7 @@ ATExecSetNotNull(List **wqueue, Relation rel, char *conName, char *colName,
 							  RelationGetRelid(rel), attnum);
 
 	/* Mark pg_attribute.attnotnull for the column */
-	set_attnotnull(wqueue, rel, attnum, lockmode);
+	set_attnotnull(wqueue, rel, attnum, constraint->skip_validation, lockmode);
 
 	/*
 	 * Recurse to propagate the constraint to children that don't have one.
@@ -9388,6 +9445,13 @@ ATPrepAddPrimaryKey(List **wqueue, Relation rel, AlterTableCmd *cmd,
 		AlterTableCmd *newcmd;
 		Constraint *nnconstr;
 
+		/* Verify that the not-null constraint has been validated */
+		if (check_for_invalid_notnull(RelationGetRelid(rel), strVal(lfirst(lc))))
+			ereport(ERROR,
+					errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+					errmsg("column \"%s\" of table \"%s\" is marked as NOT VALID NOT NULL constraint",
+						   strVal(lfirst(lc)), RelationGetRelationName(rel)));
+
 		nnconstr = makeNotNullConstraint(lfirst(lc));
 
 		newcmd = makeNode(AlterTableCmd);
@@ -9399,6 +9463,28 @@ 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 = findNotNullConstraint(relid, attname);
+	if (!HeapTupleIsValid(tuple))
+		return false;
+
+	if (((Form_pg_constraint) GETSTRUCT(tuple))->convalidated == false)
+		retval = true;
+
+	heap_freetuple(tuple);
+
+	return retval;
+}
+
 /*
  * ALTER TABLE ADD INDEX
  *
@@ -9766,7 +9852,7 @@ 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, lockmode);
 
 		ObjectAddressSet(address, ConstraintRelationId, ccon->conoid);
 	}
@@ -12367,7 +12453,8 @@ 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",
@@ -12389,6 +12476,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 +12697,114 @@ 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. call function with wqueue as NULL to
+	 * bypass validation, as it has already been performed.
+	 */
+	set_attinvalidnotnull(rel, attnum, false);
+
+	InvokeObjectPostAlterHook(ConstraintRelationId, con->oid, 0);
+
+	heap_freetuple(copyTuple);
+}
+
 /*
  * transformColumnNameList - transform list of column names
  *
diff --git a/src/backend/optimizer/util/plancat.c b/src/backend/optimizer/util/plancat.c
index 71abb01f655..5cc6980eced 100644
--- a/src/backend/optimizer/util/plancat.c
+++ b/src/backend/optimizer/util/plancat.c
@@ -28,6 +28,7 @@
 #include "catalog/catalog.h"
 #include "catalog/heap.h"
 #include "catalog/pg_am.h"
+#include "catalog/pg_constraint.h"
 #include "catalog/pg_proc.h"
 #include "catalog/pg_statistic_ext.h"
 #include "catalog/pg_statistic_ext_data.h"
@@ -177,7 +178,7 @@ get_relation_info(PlannerInfo *root, Oid relationObjectId, bool inhparent,
 		{
 			CompactAttribute *attr = TupleDescCompactAttr(relation->rd_att, i);
 
-			if (attr->attnotnull)
+			if (attr->attnotnull && !attr->attinvalidnotnull)
 			{
 				rel->notnullattnums = bms_add_member(rel->notnullattnums,
 													 i + 1);
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/bin/psql/describe.c b/src/bin/psql/describe.c
index e6cf468ac9e..b7e444f7489 100644
--- a/src/bin/psql/describe.c
+++ b/src/bin/psql/describe.c
@@ -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,14 @@ 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/access/tupdesc.h b/src/include/access/tupdesc.h
index 396eeb7a0bb..fcea828e713 100644
--- a/src/include/access/tupdesc.h
+++ b/src/include/access/tupdesc.h
@@ -45,6 +45,7 @@ typedef struct TupleConstr
 	bool		has_not_null;
 	bool		has_generated_stored;
 	bool		has_generated_virtual;
+	bool        has_invalid_not_null;
 } TupleConstr;
 
 /*
@@ -77,6 +78,7 @@ typedef struct CompactAttribute
 	bool		attisdropped;	/* as FormData_pg_attribute.attisdropped */
 	bool		attgenerated;	/* FormData_pg_attribute.attgenerated != '\0' */
 	bool		attnotnull;		/* as FormData_pg_attribute.attnotnull */
+	bool		attinvalidnotnull; /* as FormData_pg_attribute.attinvalidnotnull */
 	uint8		attalignby;		/* alignment requirement in bytes */
 } CompactAttribute;
 
diff --git a/src/include/catalog/pg_attribute.h b/src/include/catalog/pg_attribute.h
index deaa515fe53..bac4267d21d 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 "INVALID NOT NULL" 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..0e01ff1bbbd 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 4f39100fcdf..8e22a492c7d 100644
--- a/src/test/regress/expected/constraints.out
+++ b/src/test/regress/expected/constraints.out
@@ -896,6 +896,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 21ce4177de4..7aff17e6764 100644
--- a/src/test/regress/sql/constraints.sql
+++ b/src/test/regress/sql/constraints.sql
@@ -640,6 +640,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