Can't find not null constraint, but \d+ shows that

From 9c84e69b94f9a0094d224e5af3b40f51a8fac01d Mon Sep 17 00:00:00 2001
From: "tender.wang" <tender.wang@openpie.com>
Date: Wed, 27 Mar 2024 22:00:17 +0800
Subject: [PATCH v2] Fix pg_attribute attnotnull not reset when dropped PK
 indirectly.

Since b0e96f3119, we will reset pg_attribute attnotnull when drop
pk constraint and not difine NOT NULL. But in some cases, PK will
be dropped indirectly, which will not call tablecmds.c codes.

The result is \d will show not null, but we can't drop NOT NULL.
Moving some code from dropconstraint_internal to RemoveConstraintById
can fix this issue.

We should add SetAttrNotNull sub-command for some Alter Table. Because
those DDLs will check attnotnull is true during DDL.
---
 src/backend/catalog/pg_constraint.c       | 123 ++++++++++++++++++++++
 src/backend/commands/tablecmds.c          |  11 +-
 src/test/regress/expected/constraints.out |   6 ++
 src/test/regress/sql/constraints.sql      |   7 ++
 4 files changed, 142 insertions(+), 5 deletions(-)

diff --git a/src/backend/catalog/pg_constraint.c b/src/backend/catalog/pg_constraint.c
index 604280d322..8c270909bc 100644
--- a/src/backend/catalog/pg_constraint.c
+++ b/src/backend/catalog/pg_constraint.c
@@ -19,6 +19,7 @@
 #include "access/htup_details.h"
 #include "access/sysattr.h"
 #include "access/table.h"
+#include "access/xact.h"
 #include "catalog/catalog.h"
 #include "catalog/dependency.h"
 #include "catalog/heap.h"
@@ -916,6 +917,8 @@ RemoveConstraintById(Oid conId)
 	Relation	conDesc;
 	HeapTuple	tup;
 	Form_pg_constraint con;
+	List	*unconstrained_cols = NIL;
+	bool	dropping_pk = false;
 
 	conDesc = table_open(ConstraintRelationId, RowExclusiveLock);
 
@@ -967,6 +970,32 @@ RemoveConstraintById(Oid conId)
 
 			table_close(pgrel, RowExclusiveLock);
 		}
+		else if (con->contype == CONSTRAINT_PRIMARY)
+		{
+			Datum adatum;
+			ArrayType *arr;
+			int numkeys;
+			bool isNull;
+			int16 *attnums;
+
+			dropping_pk = true;
+
+			adatum = heap_getattr(tup, Anum_pg_constraint_conkey,
+								  RelationGetDescr(conDesc), &isNull);
+			if (isNull)
+				elog(ERROR, "null conkey for constraint %u", con->oid);
+			arr = DatumGetArrayTypeP(adatum); /* ensure not toasted */
+			numkeys = ARR_DIMS(arr)[0];
+			if (ARR_NDIM(arr) != 1 ||
+				numkeys < 0 ||
+				ARR_HASNULL(arr) ||
+				ARR_ELEMTYPE(arr) != INT2OID)
+				elog(ERROR, "conkey is not a 1-D smallint array");
+			attnums = (int16 *)ARR_DATA_PTR(arr);
+
+			for (int i = 0; i < numkeys; i++)
+				unconstrained_cols = lappend_int(unconstrained_cols, attnums[i]);
+		}
 
 		/* Keep lock on constraint's rel until end of xact */
 		table_close(rel, NoLock);
@@ -986,6 +1015,100 @@ RemoveConstraintById(Oid conId)
 	/* Fry the constraint itself */
 	CatalogTupleDelete(conDesc, &tup->t_self);
 
+	/*
+	 * If this was a NOT NULL or the primary key, the constrained columns must
+	 * have had pg_attribute.attnotnull set.  See if we need to reset it, and
+	 * do so.
+	 */
+	if (unconstrained_cols)
+	{
+		Relation	attrel;
+		Relation	rel;
+		Bitmapset  *pkcols;
+		Bitmapset  *ircols;
+		ListCell   *lc;
+
+		/* Make the above deletion visible */
+		CommandCounterIncrement();
+
+		attrel = table_open(AttributeRelationId, RowExclusiveLock);
+		rel = table_open(con->conrelid, RowExclusiveLock);
+
+		/*
+		 * We want to test columns for their presence in the primary key, but
+		 * only if we're not dropping it.
+		 */
+		pkcols = dropping_pk ? NULL :
+			RelationGetIndexAttrBitmap(rel,
+									   INDEX_ATTR_BITMAP_PRIMARY_KEY);
+		ircols = RelationGetIndexAttrBitmap(rel, INDEX_ATTR_BITMAP_IDENTITY_KEY);
+
+		foreach(lc, unconstrained_cols)
+		{
+			AttrNumber	attnum = lfirst_int(lc);
+			HeapTuple	atttup;
+			HeapTuple	contup;
+			Form_pg_attribute attForm;
+
+			/*
+			 * Obtain pg_attribute tuple and verify conditions on it.  We use
+			 * a copy we can scribble on.
+			 */
+			atttup = SearchSysCacheCopyAttNum(RelationGetRelid(rel), attnum);
+			if (!HeapTupleIsValid(atttup))
+				elog(ERROR, "cache lookup failed for attribute %d of relation %u",
+					 attnum, RelationGetRelid(rel));
+			attForm = (Form_pg_attribute) GETSTRUCT(atttup);
+
+			/*
+			 * Since the above deletion has been made visible, we can now
+			 * search for any remaining constraints on this column (or these
+			 * columns, in the case we're dropping a multicol primary key.)
+			 * Then, verify whether any further NOT NULL or primary key
+			 * exists, and reset attnotnull if none.
+			 *
+			 * However, if this is a generated identity column, abort the
+			 * whole thing with a specific error message, because the
+			 * constraint is required in that case.
+			 */
+			contup = findNotNullConstraintAttnum(RelationGetRelid(rel), attnum);
+			if (contup ||
+				bms_is_member(attnum - FirstLowInvalidHeapAttributeNumber,
+							  pkcols))
+				continue;
+
+			/*
+			 * It's not valid to drop the not-null constraint for a GENERATED
+			 * AS IDENTITY column.
+			 */
+			if (attForm->attidentity)
+				ereport(ERROR,
+						errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+						errmsg("column \"%s\" of relation \"%s\" is an identity column",
+							   get_attname(RelationGetRelid(rel), attnum,
+										   false),
+							   RelationGetRelationName(rel)));
+
+			/*
+			 * It's not valid to drop the not-null constraint for a column in
+			 * the replica identity index, either. (FULL is not affected.)
+			 */
+			if (bms_is_member(lfirst_int(lc) - FirstLowInvalidHeapAttributeNumber, ircols))
+				ereport(ERROR,
+						errcode(ERRCODE_INVALID_TABLE_DEFINITION),
+						errmsg("column \"%s\" is in index used as replica identity",
+							   get_attname(RelationGetRelid(rel), lfirst_int(lc), false)));
+
+			/* Reset attnotnull */
+			if (attForm->attnotnull)
+			{
+				attForm->attnotnull = false;
+				CatalogTupleUpdate(attrel, &atttup->t_self, atttup);
+			}
+		}
+		table_close(rel, RowExclusiveLock);
+		table_close(attrel, RowExclusiveLock);
+	}
 	/* Clean up */
 	ReleaseSysCache(tup);
 	table_close(conDesc, RowExclusiveLock);
diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c
index 259b4237a2..3f40f81621 100644
--- a/src/backend/commands/tablecmds.c
+++ b/src/backend/commands/tablecmds.c
@@ -147,6 +147,7 @@ typedef enum AlterTablePass
 	AT_PASS_ALTER_TYPE,			/* ALTER COLUMN TYPE */
 	AT_PASS_ADD_COL,			/* ADD COLUMN */
 	AT_PASS_SET_EXPRESSION,		/* ALTER SET EXPRESSION */
+	AT_PASS_OLD_COL_ATTRS,		/* re-set column NOT NULL */
 	AT_PASS_OLD_INDEX,			/* re-add existing indexes */
 	AT_PASS_OLD_CONSTR,			/* re-add existing constraints */
 	/* We could support a RENAME COLUMN pass here, but not currently used */
@@ -14629,12 +14630,12 @@ ATPostAlterTypeParse(Oid oldId, Oid oldRelId, Oid refRelId, char *cmd,
 				else if (cmd->subtype == AT_SetAttNotNull)
 				{
 					/*
-					 * The parser will create AT_AttSetNotNull subcommands for
-					 * columns of PRIMARY KEY indexes/constraints, but we need
-					 * not do anything with them here, because the columns'
-					 * NOT NULL marks will already have been propagated into
-					 * the new table definition.
+					 * PK drop now will reset pg_attribute attnotnull to false.
+					 * We should set attnotnull to true again.
 					 */
+					cmd->subtype = AT_SetAttNotNull;
+					tab->subcmds[AT_PASS_OLD_COL_ATTRS] =
+						lappend(tab->subcmds[AT_PASS_OLD_COL_ATTRS], cmd);
 				}
 				else
 					elog(ERROR, "unexpected statement subtype: %d",
diff --git a/src/test/regress/expected/constraints.out b/src/test/regress/expected/constraints.out
index d9d8408e86..00fab56bf8 100644
--- a/src/test/regress/expected/constraints.out
+++ b/src/test/regress/expected/constraints.out
@@ -867,6 +867,12 @@ select conname, contype, conkey from pg_constraint where conrelid = 'notnull_tbl
 (1 row)
 
 DROP TABLE notnull_tbl1;
+-- NOT NULL mask should be reset when drop PK
+CREATE TABLE notnull_t1(c0 int, c1 int);
+ALTER TABLE  notnull_t1 ADD CONSTRAINT Q PRIMARY KEY(c0, c1);
+ALTER TABLE  notnull_t1 DROP c1;
+ALTER TABLE  notnull_t1  ALTER c0 DROP NOT NULL;
+DROP TABLE notnull_t1;
 -- nope
 CREATE TABLE notnull_tbl2 (a INTEGER CONSTRAINT blah NOT NULL, b INTEGER CONSTRAINT blah NOT NULL);
 ERROR:  constraint "blah" for relation "notnull_tbl2" already exists
diff --git a/src/test/regress/sql/constraints.sql b/src/test/regress/sql/constraints.sql
index 87d685ae39..a146b67416 100644
--- a/src/test/regress/sql/constraints.sql
+++ b/src/test/regress/sql/constraints.sql
@@ -599,6 +599,13 @@ ALTER TABLE notnull_tbl1 ADD CONSTRAINT foobar NOT NULL a;
 select conname, contype, conkey from pg_constraint where conrelid = 'notnull_tbl1'::regclass;
 DROP TABLE notnull_tbl1;
 
+-- NOT NULL mask should be reset when drop PK
+CREATE TABLE notnull_t1(c0 int, c1 int);
+ALTER TABLE  notnull_t1 ADD CONSTRAINT Q PRIMARY KEY(c0, c1);
+ALTER TABLE  notnull_t1 DROP c1;
+ALTER TABLE  notnull_t1  ALTER c0 DROP NOT NULL;
+DROP TABLE notnull_t1;
+
 -- nope
 CREATE TABLE notnull_tbl2 (a INTEGER CONSTRAINT blah NOT NULL, b INTEGER CONSTRAINT blah NOT NULL);
 
-- 
2.25.1

From afb98ba63161f9e7b88aaff4b9d21d7ca64c8244 Mon Sep 17 00:00:00 2001
From: "tender.wang" <tender.wang@openpie.com>
Date: Wed, 27 Mar 2024 22:00:17 +0800
Subject: [PATCH v4] Fix pg_attribute attnotnull not reset when dropped PK
 indirectly.

Since b0e96f3119, we will reset pg_attribute attnotnull when drop
pk constraint and not difine NOT NULL. But in some cases, PK will
be dropped indirectly, which will not call tablecmds.c codes.

The result is \d will show not null, but we can't drop NOT NULL.
Moving some code from dropconstraint_internal to RemoveConstraintById
can fix this issue.

We should add SetAttrNotNull sub-command for some Alter Table. Because
those DDLs will check attnotnull is true during DDL.
---
 src/backend/catalog/pg_constraint.c       | 105 +++++++++++++++++++++-
 src/backend/commands/tablecmds.c          |  13 +--
 src/test/regress/expected/constraints.out |   6 ++
 src/test/regress/sql/constraints.sql      |   7 ++
 4 files changed, 125 insertions(+), 6 deletions(-)

diff --git a/src/backend/catalog/pg_constraint.c b/src/backend/catalog/pg_constraint.c
index 604280d322..f16f21985d 100644
--- a/src/backend/catalog/pg_constraint.c
+++ b/src/backend/catalog/pg_constraint.c
@@ -19,6 +19,7 @@
 #include "access/htup_details.h"
 #include "access/sysattr.h"
 #include "access/table.h"
+#include "access/xact.h"
 #include "catalog/catalog.h"
 #include "catalog/dependency.h"
 #include "catalog/heap.h"
@@ -906,7 +907,6 @@ RelationGetNotNullConstraints(Oid relid, bool cooked)
 	return notnulls;
 }
 
-
 /*
  * Delete a single constraint record.
  */
@@ -916,6 +916,7 @@ RemoveConstraintById(Oid conId)
 	Relation	conDesc;
 	HeapTuple	tup;
 	Form_pg_constraint con;
+	List	*unconstrained_cols = NIL;
 
 	conDesc = table_open(ConstraintRelationId, RowExclusiveLock);
 
@@ -967,6 +968,30 @@ RemoveConstraintById(Oid conId)
 
 			table_close(pgrel, RowExclusiveLock);
 		}
+		else if (con->contype == CONSTRAINT_PRIMARY)
+		{
+			Datum adatum;
+			ArrayType *arr;
+			int numkeys;
+			bool isNull;
+			int16 *attnums;
+
+			adatum = heap_getattr(tup, Anum_pg_constraint_conkey,
+								  RelationGetDescr(conDesc), &isNull);
+			if (isNull)
+				elog(ERROR, "null conkey for constraint %u", con->oid);
+			arr = DatumGetArrayTypeP(adatum); /* ensure not toasted */
+			numkeys = ARR_DIMS(arr)[0];
+			if (ARR_NDIM(arr) != 1 ||
+				numkeys < 0 ||
+				ARR_HASNULL(arr) ||
+				ARR_ELEMTYPE(arr) != INT2OID)
+				elog(ERROR, "conkey is not a 1-D smallint array");
+			attnums = (int16 *)ARR_DATA_PTR(arr);
+
+			for (int i = 0; i < numkeys; i++)
+				unconstrained_cols = lappend_int(unconstrained_cols, attnums[i]);
+		}
 
 		/* Keep lock on constraint's rel until end of xact */
 		table_close(rel, NoLock);
@@ -986,6 +1011,84 @@ RemoveConstraintById(Oid conId)
 	/* Fry the constraint itself */
 	CatalogTupleDelete(conDesc, &tup->t_self);
 
+	/*
+	 * If this was a NOT NULL or the primary key, the constrained columns must
+	 * have had pg_attribute.attnotnull set.  See if we need to reset it, and
+	 * do so.
+	 */
+	if (unconstrained_cols != NIL)
+	{
+		Relation	attrel;
+		Relation	rel;
+		Bitmapset  *ircols;
+		ListCell   *lc;
+
+		/* Make the above deletion visible */
+		CommandCounterIncrement();
+
+		attrel = table_open(AttributeRelationId, RowExclusiveLock);
+		rel = table_open(con->conrelid, AccessExclusiveLock);
+
+		ircols = RelationGetIndexAttrBitmap(rel, INDEX_ATTR_BITMAP_IDENTITY_KEY);
+
+		foreach(lc, unconstrained_cols)
+		{
+			AttrNumber	attnum = lfirst_int(lc);
+			HeapTuple	atttup;
+			HeapTuple	contup;
+			Form_pg_attribute attForm;
+
+			/*
+			 * Obtain pg_attribute tuple and verify conditions on it.  We use
+			 * a copy we can scribble on.
+			 */
+			atttup = SearchSysCacheCopyAttNum(RelationGetRelid(rel), attnum);
+			if (!HeapTupleIsValid(atttup))
+				elog(ERROR, "cache lookup failed for attribute %d of relation %u",
+					 attnum, RelationGetRelid(rel));
+			attForm = (Form_pg_attribute) GETSTRUCT(atttup);
+
+			/* GENERATED AS IDENTITY column need not null */
+			if (attForm->attidentity)
+			{
+				heap_freetuple(atttup);
+				continue;
+			}
+
+			/* A column in the replica identity index need not null */
+			if (bms_is_member(lfirst_int(lc) - FirstLowInvalidHeapAttributeNumber, ircols))
+			{
+				heap_freetuple(atttup);
+				continue;
+			}
+
+			/*
+			 * Since the above deletion has been made visible, we can now
+			 * search for any remaining constraints on this column (or these
+			 * columns, in the case we're dropping a multicol primary key.)
+			 * Then, verify whether any further NOT NULL exists, and reset
+			 * attnotnull if none.
+			 */
+			contup = findNotNullConstraintAttnum(RelationGetRelid(rel), attnum);
+			if (HeapTupleIsValid(contup))
+			{
+				heap_freetuple(contup);
+				heap_freetuple(atttup);
+				continue;
+			}
+
+			/* Reset attnotnull */
+			if (attForm->attnotnull)
+			{
+				attForm->attnotnull = false;
+				CatalogTupleUpdate(attrel, &atttup->t_self, atttup);
+			}
+
+			heap_freetuple(atttup);
+		}
+		table_close(rel, AccessExclusiveLock);
+		table_close(attrel, RowExclusiveLock);
+	}
 	/* Clean up */
 	ReleaseSysCache(tup);
 	table_close(conDesc, RowExclusiveLock);
diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c
index 259b4237a2..57fafb839a 100644
--- a/src/backend/commands/tablecmds.c
+++ b/src/backend/commands/tablecmds.c
@@ -147,6 +147,7 @@ typedef enum AlterTablePass
 	AT_PASS_ALTER_TYPE,			/* ALTER COLUMN TYPE */
 	AT_PASS_ADD_COL,			/* ADD COLUMN */
 	AT_PASS_SET_EXPRESSION,		/* ALTER SET EXPRESSION */
+	AT_PASS_OLD_COL_ATTRS,		/* re-set column NOT NULL */
 	AT_PASS_OLD_INDEX,			/* re-add existing indexes */
 	AT_PASS_OLD_CONSTR,			/* re-add existing constraints */
 	/* We could support a RENAME COLUMN pass here, but not currently used */
@@ -14629,12 +14630,14 @@ ATPostAlterTypeParse(Oid oldId, Oid oldRelId, Oid refRelId, char *cmd,
 				else if (cmd->subtype == AT_SetAttNotNull)
 				{
 					/*
-					 * The parser will create AT_AttSetNotNull subcommands for
-					 * columns of PRIMARY KEY indexes/constraints, but we need
-					 * not do anything with them here, because the columns'
-					 * NOT NULL marks will already have been propagated into
-					 * the new table definition.
+					 * When doing alter primary key column type, Removing old PK
+					 * constraint as sub-command will reset attnotnull to false in
+					 * RemoveConstraintById(). We must set attnotnull back to true
+					 * before doing re-add the index sub-command.
 					 */
+					cmd->subtype = AT_SetAttNotNull;
+					tab->subcmds[AT_PASS_OLD_COL_ATTRS] =
+						lappend(tab->subcmds[AT_PASS_OLD_COL_ATTRS], cmd);
 				}
 				else
 					elog(ERROR, "unexpected statement subtype: %d",
diff --git a/src/test/regress/expected/constraints.out b/src/test/regress/expected/constraints.out
index d9d8408e86..00fab56bf8 100644
--- a/src/test/regress/expected/constraints.out
+++ b/src/test/regress/expected/constraints.out
@@ -867,6 +867,12 @@ select conname, contype, conkey from pg_constraint where conrelid = 'notnull_tbl
 (1 row)
 
 DROP TABLE notnull_tbl1;
+-- NOT NULL mask should be reset when drop PK
+CREATE TABLE notnull_t1(c0 int, c1 int);
+ALTER TABLE  notnull_t1 ADD CONSTRAINT Q PRIMARY KEY(c0, c1);
+ALTER TABLE  notnull_t1 DROP c1;
+ALTER TABLE  notnull_t1  ALTER c0 DROP NOT NULL;
+DROP TABLE notnull_t1;
 -- nope
 CREATE TABLE notnull_tbl2 (a INTEGER CONSTRAINT blah NOT NULL, b INTEGER CONSTRAINT blah NOT NULL);
 ERROR:  constraint "blah" for relation "notnull_tbl2" already exists
diff --git a/src/test/regress/sql/constraints.sql b/src/test/regress/sql/constraints.sql
index 87d685ae39..a146b67416 100644
--- a/src/test/regress/sql/constraints.sql
+++ b/src/test/regress/sql/constraints.sql
@@ -599,6 +599,13 @@ ALTER TABLE notnull_tbl1 ADD CONSTRAINT foobar NOT NULL a;
 select conname, contype, conkey from pg_constraint where conrelid = 'notnull_tbl1'::regclass;
 DROP TABLE notnull_tbl1;
 
+-- NOT NULL mask should be reset when drop PK
+CREATE TABLE notnull_t1(c0 int, c1 int);
+ALTER TABLE  notnull_t1 ADD CONSTRAINT Q PRIMARY KEY(c0, c1);
+ALTER TABLE  notnull_t1 DROP c1;
+ALTER TABLE  notnull_t1  ALTER c0 DROP NOT NULL;
+DROP TABLE notnull_t1;
+
 -- nope
 CREATE TABLE notnull_tbl2 (a INTEGER CONSTRAINT blah NOT NULL, b INTEGER CONSTRAINT blah NOT NULL);
 
-- 
2.25.1

From 397d424cd64020bcb4aff219deefca7fc7b6f88d Mon Sep 17 00:00:00 2001
From: Alvaro Herrera <alvherre@alvh.no-ip.org>
Date: Tue, 9 Apr 2024 19:18:57 +0200
Subject: [PATCH] Correctly reset attnotnull when constraints dropped
 indirectly

---
 src/backend/catalog/pg_constraint.c       | 105 +++++++++++++++++++++-
 src/backend/commands/tablecmds.c          |  67 ++++++--------
 src/test/regress/expected/constraints.out |  11 +++
 src/test/regress/sql/constraints.sql      |   7 ++
 4 files changed, 150 insertions(+), 40 deletions(-)

diff --git a/src/backend/catalog/pg_constraint.c b/src/backend/catalog/pg_constraint.c
index 604280d322..17c7a2d0f3 100644
--- a/src/backend/catalog/pg_constraint.c
+++ b/src/backend/catalog/pg_constraint.c
@@ -19,6 +19,7 @@
 #include "access/htup_details.h"
 #include "access/sysattr.h"
 #include "access/table.h"
+#include "access/xact.h"
 #include "catalog/catalog.h"
 #include "catalog/dependency.h"
 #include "catalog/heap.h"
@@ -916,6 +917,8 @@ RemoveConstraintById(Oid conId)
 	Relation	conDesc;
 	HeapTuple	tup;
 	Form_pg_constraint con;
+	bool		dropping_pk = false;
+	List	   *unconstrained_cols = NIL;
 
 	conDesc = table_open(ConstraintRelationId, RowExclusiveLock);
 
@@ -940,7 +943,9 @@ RemoveConstraintById(Oid conId)
 		/*
 		 * We need to update the relchecks count if it is a check constraint
 		 * being dropped.  This update will force backends to rebuild relcache
-		 * entries when we commit.
+		 * entries when we commit.  For not-null and primary key constraints,
+		 * obtain the list of columns affected, so that we can reset their
+		 * attnotnull flags below.
 		 */
 		if (con->contype == CONSTRAINT_CHECK)
 		{
@@ -967,6 +972,36 @@ RemoveConstraintById(Oid conId)
 
 			table_close(pgrel, RowExclusiveLock);
 		}
+		else if (con->contype == CONSTRAINT_NOTNULL)
+		{
+			unconstrained_cols = list_make1_int(extractNotNullColumn(tup));
+		}
+		else if (con->contype == CONSTRAINT_PRIMARY)
+		{
+			Datum		adatum;
+			ArrayType  *arr;
+			int			numkeys;
+			bool		isNull;
+			int16	   *attnums;
+
+			dropping_pk = true;
+
+			adatum = heap_getattr(tup, Anum_pg_constraint_conkey,
+								  RelationGetDescr(conDesc), &isNull);
+			if (isNull)
+				elog(ERROR, "null conkey for constraint %u", con->oid);
+			arr = DatumGetArrayTypeP(adatum);	/* ensure not toasted */
+			numkeys = ARR_DIMS(arr)[0];
+			if (ARR_NDIM(arr) != 1 ||
+				numkeys < 0 ||
+				ARR_HASNULL(arr) ||
+				ARR_ELEMTYPE(arr) != INT2OID)
+				elog(ERROR, "conkey is not a 1-D smallint array");
+			attnums = (int16 *) ARR_DATA_PTR(arr);
+
+			for (int i = 0; i < numkeys; i++)
+				unconstrained_cols = lappend_int(unconstrained_cols, attnums[i]);
+		}
 
 		/* Keep lock on constraint's rel until end of xact */
 		table_close(rel, NoLock);
@@ -986,6 +1021,74 @@ RemoveConstraintById(Oid conId)
 	/* Fry the constraint itself */
 	CatalogTupleDelete(conDesc, &tup->t_self);
 
+	/*
+	 * If this was a NOT NULL or the primary key, the constrained columns must
+	 * have had pg_attribute.attnotnull set.  See if we need to reset it, and
+	 * do so.
+	 */
+	if (unconstrained_cols)
+	{
+		Relation	tablerel;
+		Relation	attrel;
+		Bitmapset  *pkcols;
+		ListCell   *lc;
+
+		/* Make the above deletion visible */
+		CommandCounterIncrement();
+
+		tablerel = table_open(con->conrelid, NoLock);	/* already have lock */
+		attrel = table_open(AttributeRelationId, RowExclusiveLock);
+
+		/*
+		 * We want to test columns for their presence in the primary key, but
+		 * only if we're not dropping it.
+		 */
+		pkcols = dropping_pk ? NULL :
+			RelationGetIndexAttrBitmap(tablerel,
+									   INDEX_ATTR_BITMAP_PRIMARY_KEY);
+
+		foreach(lc, unconstrained_cols)
+		{
+			AttrNumber	attnum = lfirst_int(lc);
+			HeapTuple	atttup;
+			HeapTuple	contup;
+			Form_pg_attribute attForm;
+
+			/*
+			 * Obtain pg_attribute tuple and verify conditions on it.  We use
+			 * a copy we can scribble on.
+			 */
+			atttup = SearchSysCacheCopyAttNum(con->conrelid, attnum);
+			if (!HeapTupleIsValid(atttup))
+				elog(ERROR, "cache lookup failed for attribute %d of relation %u",
+					 attnum, con->conrelid);
+			attForm = (Form_pg_attribute) GETSTRUCT(atttup);
+
+			/*
+			 * Since the above deletion has been made visible, we can now
+			 * search for any remaining constraints on this column (or these
+			 * columns, in the case we're dropping a multicol primary key.)
+			 * Then, verify whether any further NOT NULL or primary key
+			 * exists, and reset attnotnull if none.
+			 */
+			contup = findNotNullConstraintAttnum(con->conrelid, attnum);
+			if (contup ||
+				bms_is_member(attnum - FirstLowInvalidHeapAttributeNumber,
+							  pkcols))
+				continue;
+
+			/* Reset attnotnull */
+			if (attForm->attnotnull)
+			{
+				attForm->attnotnull = false;
+				CatalogTupleUpdate(attrel, &atttup->t_self, atttup);
+			}
+		}
+
+		table_close(attrel, RowExclusiveLock);
+		table_close(tablerel, NoLock);
+	}
+
 	/* Clean up */
 	ReleaseSysCache(tup);
 	table_close(conDesc, RowExclusiveLock);
diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c
index 865c6331c1..25d3613fe4 100644
--- a/src/backend/commands/tablecmds.c
+++ b/src/backend/commands/tablecmds.c
@@ -149,6 +149,7 @@ typedef enum AlterTablePass
 	AT_PASS_ALTER_TYPE,			/* ALTER COLUMN TYPE */
 	AT_PASS_ADD_COL,			/* ADD COLUMN */
 	AT_PASS_SET_EXPRESSION,		/* ALTER SET EXPRESSION */
+	AT_PASS_OLD_COL_ATTRS,		/* re-install attnotnull */
 	AT_PASS_OLD_INDEX,			/* re-add existing indexes */
 	AT_PASS_OLD_CONSTR,			/* re-add existing constraints */
 	/* We could support a RENAME COLUMN pass here, but not currently used */
@@ -12933,10 +12934,10 @@ dropconstraint_internal(Relation rel, HeapTuple constraintTup, DropBehavior beha
 	List	   *children;
 	ListCell   *child;
 	bool		is_no_inherit_constraint = false;
-	bool		dropping_pk = false;
 	char	   *constrName;
 	List	   *unconstrained_cols = NIL;
-	char	   *colname;
+	char	   *colname = NULL;
+	bool		dropping_pk = false;
 
 	if (list_member_oid(*readyRels, RelationGetRelid(rel)))
 		return InvalidObjectAddress;
@@ -12968,10 +12969,12 @@ dropconstraint_internal(Relation rel, HeapTuple constraintTup, DropBehavior beha
 	 */
 	if (con->contype == CONSTRAINT_NOTNULL)
 	{
-		AttrNumber	colnum = extractNotNullColumn(constraintTup);
+		AttrNumber	colnum;
 
-		if (colnum != InvalidAttrNumber)
-			unconstrained_cols = list_make1_int(colnum);
+		colnum = extractNotNullColumn(constraintTup);
+		unconstrained_cols = list_make1_int(colnum);
+		colname = NameStr(TupleDescAttr(RelationGetDescr(rel),
+										colnum - 1)->attname);
 	}
 	else if (con->contype == CONSTRAINT_PRIMARY)
 	{
@@ -13027,9 +13030,8 @@ dropconstraint_internal(Relation rel, HeapTuple constraintTup, DropBehavior beha
 	performDeletion(&conobj, behavior, 0);
 
 	/*
-	 * If this was a NOT NULL or the primary key, the constrained columns must
-	 * have had pg_attribute.attnotnull set.  See if we need to reset it, and
-	 * do so.
+	 * If this was a NOT NULL or the primary key, verify that we still have
+	 * constraints to support GENERATED AS IDENTITY or the replica identity.
 	 */
 	if (unconstrained_cols)
 	{
@@ -13038,7 +13040,7 @@ dropconstraint_internal(Relation rel, HeapTuple constraintTup, DropBehavior beha
 		Bitmapset  *ircols;
 		ListCell   *lc;
 
-		/* Make the above deletion visible */
+		/* Make implicit attnotnull changes visible */
 		CommandCounterIncrement();
 
 		attrel = table_open(AttributeRelationId, RowExclusiveLock);
@@ -13058,27 +13060,26 @@ dropconstraint_internal(Relation rel, HeapTuple constraintTup, DropBehavior beha
 			HeapTuple	atttup;
 			HeapTuple	contup;
 			Form_pg_attribute attForm;
+			char		attidentity;
 
 			/*
-			 * Obtain pg_attribute tuple and verify conditions on it.  We use
-			 * a copy we can scribble on.
+			 * Obtain pg_attribute tuple and verify conditions on it.
 			 */
-			atttup = SearchSysCacheCopyAttNum(RelationGetRelid(rel), attnum);
+			atttup = SearchSysCacheAttNum(RelationGetRelid(rel), attnum);
 			if (!HeapTupleIsValid(atttup))
 				elog(ERROR, "cache lookup failed for attribute %d of relation %u",
 					 attnum, RelationGetRelid(rel));
 			attForm = (Form_pg_attribute) GETSTRUCT(atttup);
+			attidentity = attForm->attidentity;
+			ReleaseSysCache(atttup);
 
 			/*
 			 * Since the above deletion has been made visible, we can now
 			 * search for any remaining constraints on this column (or these
 			 * columns, in the case we're dropping a multicol primary key.)
 			 * Then, verify whether any further NOT NULL or primary key
-			 * exists, and reset attnotnull if none.
-			 *
-			 * However, if this is a generated identity column, abort the
-			 * whole thing with a specific error message, because the
-			 * constraint is required in that case.
+			 * exists: if none and this is a generated identity column or the
+			 * replica identity, abort the whole thing.
 			 */
 			contup = findNotNullConstraintAttnum(RelationGetRelid(rel), attnum);
 			if (contup ||
@@ -13090,7 +13091,7 @@ dropconstraint_internal(Relation rel, HeapTuple constraintTup, DropBehavior beha
 			 * It's not valid to drop the not-null constraint for a GENERATED
 			 * AS IDENTITY column.
 			 */
-			if (attForm->attidentity)
+			if (attidentity != '\0')
 				ereport(ERROR,
 						errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
 						errmsg("column \"%s\" of relation \"%s\" is an identity column",
@@ -13107,13 +13108,6 @@ dropconstraint_internal(Relation rel, HeapTuple constraintTup, DropBehavior beha
 						errcode(ERRCODE_INVALID_TABLE_DEFINITION),
 						errmsg("column \"%s\" is in index used as replica identity",
 							   get_attname(RelationGetRelid(rel), lfirst_int(lc), false)));
-
-			/* Reset attnotnull */
-			if (attForm->attnotnull)
-			{
-				attForm->attnotnull = false;
-				CatalogTupleUpdate(attrel, &atttup->t_self, atttup);
-			}
 		}
 		table_close(attrel, RowExclusiveLock);
 	}
@@ -13152,13 +13146,6 @@ dropconstraint_internal(Relation rel, HeapTuple constraintTup, DropBehavior beha
 				 errmsg("cannot remove constraint from only the partitioned table when partitions exist"),
 				 errhint("Do not specify the ONLY keyword.")));
 
-	/* For not-null constraints we recurse by column name */
-	if (con->contype == CONSTRAINT_NOTNULL)
-		colname = NameStr(TupleDescAttr(RelationGetDescr(rel),
-										linitial_int(unconstrained_cols) - 1)->attname);
-	else
-		colname = NULL;			/* keep compiler quiet */
-
 	foreach(child, children)
 	{
 		Oid			childrelid = lfirst_oid(child);
@@ -13174,8 +13161,8 @@ dropconstraint_internal(Relation rel, HeapTuple constraintTup, DropBehavior beha
 		CheckTableNotInUse(childrel, "ALTER TABLE");
 
 		/*
-		 * We search for not-null constraint by column number, and other
-		 * constraints by name.
+		 * We search for not-null constraints by column name, and others by
+		 * constraint name.
 		 */
 		if (con->contype == CONSTRAINT_NOTNULL)
 		{
@@ -14656,12 +14643,14 @@ ATPostAlterTypeParse(Oid oldId, Oid oldRelId, Oid refRelId, char *cmd,
 				else if (cmd->subtype == AT_SetAttNotNull)
 				{
 					/*
-					 * The parser will create AT_AttSetNotNull subcommands for
-					 * columns of PRIMARY KEY indexes/constraints, but we need
-					 * not do anything with them here, because the columns'
-					 * NOT NULL marks will already have been propagated into
-					 * the new table definition.
+					 * When doing alter primary key column type, Removing old
+					 * PK constraint as sub-command will reset attnotnull to
+					 * false in RemoveConstraintById(). We must set attnotnull
+					 * back to true before doing re-add the index sub-command.
 					 */
+					cmd->subtype = AT_SetAttNotNull;
+					tab->subcmds[AT_PASS_OLD_COL_ATTRS] =
+						lappend(tab->subcmds[AT_PASS_OLD_COL_ATTRS], cmd);
 				}
 				else
 					elog(ERROR, "unexpected statement subtype: %d",
diff --git a/src/test/regress/expected/constraints.out b/src/test/regress/expected/constraints.out
index d9d8408e86..ce8ba73a87 100644
--- a/src/test/regress/expected/constraints.out
+++ b/src/test/regress/expected/constraints.out
@@ -866,6 +866,17 @@ select conname, contype, conkey from pg_constraint where conrelid = 'notnull_tbl
  foobar  | n       | {1}
 (1 row)
 
+DROP TABLE notnull_tbl1;
+-- make sure attnotnull is reset correctly when a PK is dropped indirectly
+CREATE TABLE notnull_tbl1 (c0 int, c1 int);
+ALTER TABLE  notnull_tbl1 ADD CONSTRAINT Q PRIMARY KEY(c0, c1);
+ALTER TABLE  notnull_tbl1 DROP c1;
+\d+ notnull_tbl1
+                               Table "public.notnull_tbl1"
+ Column |  Type   | Collation | Nullable | Default | Storage | Stats target | Description 
+--------+---------+-----------+----------+---------+---------+--------------+-------------
+ c0     | integer |           |          |         | plain   |              | 
+
 DROP TABLE notnull_tbl1;
 -- nope
 CREATE TABLE notnull_tbl2 (a INTEGER CONSTRAINT blah NOT NULL, b INTEGER CONSTRAINT blah NOT NULL);
diff --git a/src/test/regress/sql/constraints.sql b/src/test/regress/sql/constraints.sql
index 87d685ae39..b25107da46 100644
--- a/src/test/regress/sql/constraints.sql
+++ b/src/test/regress/sql/constraints.sql
@@ -599,6 +599,13 @@ ALTER TABLE notnull_tbl1 ADD CONSTRAINT foobar NOT NULL a;
 select conname, contype, conkey from pg_constraint where conrelid = 'notnull_tbl1'::regclass;
 DROP TABLE notnull_tbl1;
 
+-- make sure attnotnull is reset correctly when a PK is dropped indirectly
+CREATE TABLE notnull_tbl1 (c0 int, c1 int);
+ALTER TABLE  notnull_tbl1 ADD CONSTRAINT Q PRIMARY KEY(c0, c1);
+ALTER TABLE  notnull_tbl1 DROP c1;
+\d+ notnull_tbl1
+DROP TABLE notnull_tbl1;
+
 -- nope
 CREATE TABLE notnull_tbl2 (a INTEGER CONSTRAINT blah NOT NULL, b INTEGER CONSTRAINT blah NOT NULL);
 
-- 
2.39.2

From 3a00358847f1792c01e608909dc8a4a0edb8739c Mon Sep 17 00:00:00 2001
From: Alvaro Herrera <alvherre@alvh.no-ip.org>
Date: Wed, 10 Apr 2024 13:02:35 +0200
Subject: [PATCH] Handle ALTER .. DROP NOT NULL when no pg_constraint row
 exists

---
 src/backend/commands/tablecmds.c          | 67 +++++++++++++++++++----
 src/test/regress/expected/constraints.out | 55 +++++++++++++++++++
 src/test/regress/sql/constraints.sql      | 29 ++++++++++
 3 files changed, 139 insertions(+), 12 deletions(-)

diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c
index 865c6331c1..f692001e55 100644
--- a/src/backend/commands/tablecmds.c
+++ b/src/backend/commands/tablecmds.c
@@ -448,6 +448,7 @@ static ObjectAddress ATExecDropNotNull(Relation rel, const char *colName, bool r
 									   LOCKMODE lockmode);
 static bool set_attnotnull(List **wqueue, Relation rel,
 						   AttrNumber attnum, bool recurse, LOCKMODE lockmode);
+static void drop_orphaned_notnull(Oid relationOid, AttrNumber attnum);
 static ObjectAddress ATExecSetNotNull(List **wqueue, Relation rel,
 									  char *constrname, char *colName,
 									  bool recurse, bool recursing,
@@ -7678,17 +7679,23 @@ ATExecDropNotNull(Relation rel, const char *colName, bool recurse,
 	}
 
 	/*
-	 * Find the constraint that makes this column NOT NULL.
+	 * Find the constraint that makes this column NOT NULL, and drop it if we
+	 * see one.  dropconstraint_internal() will do necessary consistency
+	 * checking.  If there isn't one, there are two possibilities: either the
+	 * column is marked attnotnull because it's part of the primary key, and
+	 * then we just throw an appropriate error; or it's a leftover marking that
+	 * we can remove.  However, before doing the latter, to avoid breaking
+	 * consistency any further, prevent this if the column is part of the
+	 * replica identity.
 	 */
 	conTup = findNotNullConstraint(RelationGetRelid(rel), colName);
 	if (conTup == NULL)
 	{
 		Bitmapset  *pkcols;
+		Bitmapset  *ircols;
 
 		/*
-		 * There's no not-null constraint, so throw an error.  If the column
-		 * is in a primary key, we can throw a specific error.  Otherwise,
-		 * this is unexpected.
+		 * If the column is in a primary key, throw a specific error message.
 		 */
 		pkcols = RelationGetIndexAttrBitmap(rel, INDEX_ATTR_BITMAP_PRIMARY_KEY);
 		if (bms_is_member(attnum - FirstLowInvalidHeapAttributeNumber,
@@ -7697,16 +7704,25 @@ ATExecDropNotNull(Relation rel, const char *colName, bool recurse,
 					errcode(ERRCODE_INVALID_TABLE_DEFINITION),
 					errmsg("column \"%s\" is in a primary key", colName));
 
-		/* this shouldn't happen */
-		elog(ERROR, "could not find not-null constraint on column \"%s\", relation \"%s\"",
-			 colName, RelationGetRelationName(rel));
+		/* Also throw an error if the column is in the replica identity */
+		ircols = RelationGetIndexAttrBitmap(rel, INDEX_ATTR_BITMAP_IDENTITY_KEY);
+		if (bms_is_member(attnum - FirstLowInvalidHeapAttributeNumber, ircols))
+			ereport(ERROR,
+					errcode(ERRCODE_INVALID_TABLE_DEFINITION),
+					errmsg("column \"%s\" is in index used as replica identity",
+						   get_attname(RelationGetRelid(rel), attnum, false)));
+
+		/* Otherwise, just remove the attnotnull marking and do nothing else. */
+		drop_orphaned_notnull(RelationGetRelid(rel), attnum);
 	}
+	else
+	{
+		readyRels = NIL;
+		dropconstraint_internal(rel, conTup, DROP_RESTRICT, recurse, false,
+								false, &readyRels, lockmode);
 
-	readyRels = NIL;
-	dropconstraint_internal(rel, conTup, DROP_RESTRICT, recurse, false,
-							false, &readyRels, lockmode);
-
-	heap_freetuple(conTup);
+		heap_freetuple(conTup);
+	}
 
 	InvokeObjectPostAlterHook(RelationRelationId,
 							  RelationGetRelid(rel), attnum);
@@ -7796,6 +7812,33 @@ set_attnotnull(List **wqueue, Relation rel, AttrNumber attnum, bool recurse,
 	return retval;
 }
 
+/*
+ * In rare cases, a column can end up with an "orphaned" attnotnull marking
+ * with no corresponding pg_constraint row.  If the user then does ALTER TABLE
+ * DROP NOT NULL, this takes care of resetting that.
+ */
+static void
+drop_orphaned_notnull(Oid relationOid, AttrNumber attnum)
+{
+	Relation	attr_rel;
+	HeapTuple	tuple;
+	Form_pg_attribute attForm;
+
+	attr_rel = table_open(AttributeRelationId, RowExclusiveLock);
+
+	tuple = SearchSysCacheCopyAttNum(relationOid, attnum);
+	if (!HeapTupleIsValid(tuple))
+		elog(ERROR, "cache lookup failed for attribute %d of relation %u",
+			 attnum, relationOid);
+	attForm = (Form_pg_attribute) GETSTRUCT(tuple);
+
+	attForm->attnotnull = false;
+	CatalogTupleUpdate(attr_rel, &tuple->t_self, tuple);
+
+	table_close(attr_rel, RowExclusiveLock);
+}
+
+
 /*
  * ALTER TABLE ALTER COLUMN SET NOT NULL
  *
diff --git a/src/test/regress/expected/constraints.out b/src/test/regress/expected/constraints.out
index d9d8408e86..f0c166dc4f 100644
--- a/src/test/regress/expected/constraints.out
+++ b/src/test/regress/expected/constraints.out
@@ -866,6 +866,61 @@ select conname, contype, conkey from pg_constraint where conrelid = 'notnull_tbl
  foobar  | n       | {1}
 (1 row)
 
+DROP TABLE notnull_tbl1;
+-- make sure attnotnull is reset correctly when a PK is dropped indirectly
+CREATE TABLE notnull_tbl1 (c0 int, c1 int, PRIMARY KEY (c0, c1));
+ALTER TABLE  notnull_tbl1 DROP c1;
+\d+ notnull_tbl1
+                               Table "public.notnull_tbl1"
+ Column |  Type   | Collation | Nullable | Default | Storage | Stats target | Description 
+--------+---------+-----------+----------+---------+---------+--------------+-------------
+ c0     | integer |           | not null |         | plain   |              | 
+
+ALTER TABLE  notnull_tbl1 ALTER c0 DROP NOT NULL;
+\d+ notnull_tbl1
+                               Table "public.notnull_tbl1"
+ Column |  Type   | Collation | Nullable | Default | Storage | Stats target | Description 
+--------+---------+-----------+----------+---------+---------+--------------+-------------
+ c0     | integer |           |          |         | plain   |              | 
+
+DROP TABLE notnull_tbl1;
+-- again
+CREATE DOMAIN notnull_dom1 AS INTEGER;
+CREATE TABLE notnull_tbl1 (c0 notnull_dom1, c1 int, PRIMARY KEY (c0, c1));
+DROP DOMAIN notnull_dom1 CASCADE;
+NOTICE:  drop cascades to column c0 of table notnull_tbl1
+\d+ notnull_tbl1
+                               Table "public.notnull_tbl1"
+ Column |  Type   | Collation | Nullable | Default | Storage | Stats target | Description 
+--------+---------+-----------+----------+---------+---------+--------------+-------------
+ c1     | integer |           | not null |         | plain   |              | 
+
+ALTER TABLE  notnull_tbl1 ALTER c1 SET NOT NULL;
+\d+ notnull_tbl1
+                               Table "public.notnull_tbl1"
+ Column |  Type   | Collation | Nullable | Default | Storage | Stats target | Description 
+--------+---------+-----------+----------+---------+---------+--------------+-------------
+ c1     | integer |           | not null |         | plain   |              | 
+Not-null constraints:
+    "notnull_tbl1_c1_not_null" NOT NULL "c1"
+
+DROP TABLE notnull_tbl1;
+-- with a REPLICA IDENTITY column
+CREATE DOMAIN notnull_dom1 AS INTEGER;
+CREATE TABLE notnull_tbl1 (c0 notnull_dom1, c1 int UNIQUE, PRIMARY KEY (c0, c1));
+ALTER TABLE notnull_tbl1 REPLICA IDENTITY USING INDEX notnull_tbl1_c1_key;
+DROP DOMAIN notnull_dom1 CASCADE;
+NOTICE:  drop cascades to column c0 of table notnull_tbl1
+ALTER TABLE  notnull_tbl1 ALTER c1 DROP NOT NULL;
+ERROR:  column "c1" is in index used as replica identity
+DROP TABLE notnull_tbl1;
+-- with an identity generated column
+CREATE DOMAIN notnull_dom1 AS INTEGER;
+CREATE TABLE notnull_tbl1 (c0 notnull_dom1, c1 int generated by default as identity, PRIMARY KEY (c0, c1));
+DROP DOMAIN notnull_dom1 CASCADE;
+NOTICE:  drop cascades to column c0 of table notnull_tbl1
+ALTER TABLE  notnull_tbl1 ALTER c1 DROP NOT NULL;
+ERROR:  column "c1" of relation "notnull_tbl1" is an identity column
 DROP TABLE notnull_tbl1;
 -- nope
 CREATE TABLE notnull_tbl2 (a INTEGER CONSTRAINT blah NOT NULL, b INTEGER CONSTRAINT blah NOT NULL);
diff --git a/src/test/regress/sql/constraints.sql b/src/test/regress/sql/constraints.sql
index 87d685ae39..e946c29f18 100644
--- a/src/test/regress/sql/constraints.sql
+++ b/src/test/regress/sql/constraints.sql
@@ -599,6 +599,35 @@ ALTER TABLE notnull_tbl1 ADD CONSTRAINT foobar NOT NULL a;
 select conname, contype, conkey from pg_constraint where conrelid = 'notnull_tbl1'::regclass;
 DROP TABLE notnull_tbl1;
 
+-- make sure attnotnull is reset correctly when a PK is dropped indirectly
+CREATE TABLE notnull_tbl1 (c0 int, c1 int, PRIMARY KEY (c0, c1));
+ALTER TABLE  notnull_tbl1 DROP c1;
+\d+ notnull_tbl1
+ALTER TABLE  notnull_tbl1 ALTER c0 DROP NOT NULL;
+\d+ notnull_tbl1
+DROP TABLE notnull_tbl1;
+-- again
+CREATE DOMAIN notnull_dom1 AS INTEGER;
+CREATE TABLE notnull_tbl1 (c0 notnull_dom1, c1 int, PRIMARY KEY (c0, c1));
+DROP DOMAIN notnull_dom1 CASCADE;
+\d+ notnull_tbl1
+ALTER TABLE  notnull_tbl1 ALTER c1 SET NOT NULL;
+\d+ notnull_tbl1
+DROP TABLE notnull_tbl1;
+-- with a REPLICA IDENTITY column
+CREATE DOMAIN notnull_dom1 AS INTEGER;
+CREATE TABLE notnull_tbl1 (c0 notnull_dom1, c1 int UNIQUE, PRIMARY KEY (c0, c1));
+ALTER TABLE notnull_tbl1 REPLICA IDENTITY USING INDEX notnull_tbl1_c1_key;
+DROP DOMAIN notnull_dom1 CASCADE;
+ALTER TABLE  notnull_tbl1 ALTER c1 DROP NOT NULL;
+DROP TABLE notnull_tbl1;
+-- with an identity generated column
+CREATE DOMAIN notnull_dom1 AS INTEGER;
+CREATE TABLE notnull_tbl1 (c0 notnull_dom1, c1 int generated by default as identity, PRIMARY KEY (c0, c1));
+DROP DOMAIN notnull_dom1 CASCADE;
+ALTER TABLE  notnull_tbl1 ALTER c1 DROP NOT NULL;
+DROP TABLE notnull_tbl1;
+
 -- nope
 CREATE TABLE notnull_tbl2 (a INTEGER CONSTRAINT blah NOT NULL, b INTEGER CONSTRAINT blah NOT NULL);
 
-- 
2.39.2

From dd41bf8d1f2dea5725909150609f8e565e11efcf Mon Sep 17 00:00:00 2001
From: jian he <jian.universality@gmail.com>
Date: Wed, 10 Apr 2024 20:49:04 +0800
Subject: [PATCH v1 1/1] minor coesmetuic refactor in dropconstraint_internal

mainly change `foreach` to `foreach_int` while iterating through list unconstrained_cols.
---
 src/backend/commands/tablecmds.c | 14 ++++++--------
 1 file changed, 6 insertions(+), 8 deletions(-)

diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c
index 8a02c5b0..8dc623e8 100644
--- a/src/backend/commands/tablecmds.c
+++ b/src/backend/commands/tablecmds.c
@@ -12968,12 +12968,11 @@ dropconstraint_internal(Relation rel, HeapTuple constraintTup, DropBehavior beha
 	 * have had pg_attribute.attnotnull set.  See if we need to reset it, and
 	 * do so.
 	 */
-	if (unconstrained_cols)
+	if (unconstrained_cols != NIL)
 	{
 		Relation	attrel;
 		Bitmapset  *pkcols;
 		Bitmapset  *ircols;
-		ListCell   *lc;
 
 		/* Make the above deletion visible */
 		CommandCounterIncrement();
@@ -12989,9 +12988,8 @@ dropconstraint_internal(Relation rel, HeapTuple constraintTup, DropBehavior beha
 									   INDEX_ATTR_BITMAP_PRIMARY_KEY);
 		ircols = RelationGetIndexAttrBitmap(rel, INDEX_ATTR_BITMAP_IDENTITY_KEY);
 
-		foreach(lc, unconstrained_cols)
+		foreach_int(attnum, unconstrained_cols)
 		{
-			AttrNumber	attnum = lfirst_int(lc);
 			HeapTuple	atttup;
 			HeapTuple	contup;
 			Form_pg_attribute attForm;
@@ -13039,11 +13037,11 @@ dropconstraint_internal(Relation rel, HeapTuple constraintTup, DropBehavior beha
 			 * It's not valid to drop the not-null constraint for a column in
 			 * the replica identity index, either. (FULL is not affected.)
 			 */
-			if (bms_is_member(lfirst_int(lc) - FirstLowInvalidHeapAttributeNumber, ircols))
+			if (bms_is_member(attnum - FirstLowInvalidHeapAttributeNumber, ircols))
 				ereport(ERROR,
 						errcode(ERRCODE_INVALID_TABLE_DEFINITION),
 						errmsg("column \"%s\" is in index used as replica identity",
-							   get_attname(RelationGetRelid(rel), lfirst_int(lc), false)));
+							   get_attname(RelationGetRelid(rel), attnum, false)));
 
 			/* Reset attnotnull */
 			if (attForm->attnotnull)
@@ -13233,10 +13231,10 @@ dropconstraint_internal(Relation rel, HeapTuple constraintTup, DropBehavior beha
 		 * Find out the list of column names to process.  Fortunately, we
 		 * already have the list of column numbers.
 		 */
-		foreach(lc, unconstrained_cols)
+		foreach_int(attnum, unconstrained_cols)
 		{
 			colnames = lappend(colnames, get_attname(RelationGetRelid(rel),
-													 lfirst_int(lc), false));
+													 attnum, false));
 		}
 
 		foreach(child, children)
-- 
2.34.1

From 3a00358847f1792c01e608909dc8a4a0edb8739c Mon Sep 17 00:00:00 2001
From: Alvaro Herrera <alvherre@alvh.no-ip.org>
Date: Wed, 10 Apr 2024 13:02:35 +0200
Subject: [PATCH v2 1/2] Handle ALTER .. DROP NOT NULL when no pg_constraint
 row exists

---
 src/backend/commands/tablecmds.c          | 67 +++++++++++++++++++----
 src/test/regress/expected/constraints.out | 55 +++++++++++++++++++
 src/test/regress/sql/constraints.sql      | 29 ++++++++++
 3 files changed, 139 insertions(+), 12 deletions(-)

diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c
index 865c6331c1..f692001e55 100644
--- a/src/backend/commands/tablecmds.c
+++ b/src/backend/commands/tablecmds.c
@@ -448,6 +448,7 @@ static ObjectAddress ATExecDropNotNull(Relation rel, const char *colName, bool r
 									   LOCKMODE lockmode);
 static bool set_attnotnull(List **wqueue, Relation rel,
 						   AttrNumber attnum, bool recurse, LOCKMODE lockmode);
+static void drop_orphaned_notnull(Oid relationOid, AttrNumber attnum);
 static ObjectAddress ATExecSetNotNull(List **wqueue, Relation rel,
 									  char *constrname, char *colName,
 									  bool recurse, bool recursing,
@@ -7678,17 +7679,23 @@ ATExecDropNotNull(Relation rel, const char *colName, bool recurse,
 	}
 
 	/*
-	 * Find the constraint that makes this column NOT NULL.
+	 * Find the constraint that makes this column NOT NULL, and drop it if we
+	 * see one.  dropconstraint_internal() will do necessary consistency
+	 * checking.  If there isn't one, there are two possibilities: either the
+	 * column is marked attnotnull because it's part of the primary key, and
+	 * then we just throw an appropriate error; or it's a leftover marking that
+	 * we can remove.  However, before doing the latter, to avoid breaking
+	 * consistency any further, prevent this if the column is part of the
+	 * replica identity.
 	 */
 	conTup = findNotNullConstraint(RelationGetRelid(rel), colName);
 	if (conTup == NULL)
 	{
 		Bitmapset  *pkcols;
+		Bitmapset  *ircols;
 
 		/*
-		 * There's no not-null constraint, so throw an error.  If the column
-		 * is in a primary key, we can throw a specific error.  Otherwise,
-		 * this is unexpected.
+		 * If the column is in a primary key, throw a specific error message.
 		 */
 		pkcols = RelationGetIndexAttrBitmap(rel, INDEX_ATTR_BITMAP_PRIMARY_KEY);
 		if (bms_is_member(attnum - FirstLowInvalidHeapAttributeNumber,
@@ -7697,16 +7704,25 @@ ATExecDropNotNull(Relation rel, const char *colName, bool recurse,
 					errcode(ERRCODE_INVALID_TABLE_DEFINITION),
 					errmsg("column \"%s\" is in a primary key", colName));
 
-		/* this shouldn't happen */
-		elog(ERROR, "could not find not-null constraint on column \"%s\", relation \"%s\"",
-			 colName, RelationGetRelationName(rel));
+		/* Also throw an error if the column is in the replica identity */
+		ircols = RelationGetIndexAttrBitmap(rel, INDEX_ATTR_BITMAP_IDENTITY_KEY);
+		if (bms_is_member(attnum - FirstLowInvalidHeapAttributeNumber, ircols))
+			ereport(ERROR,
+					errcode(ERRCODE_INVALID_TABLE_DEFINITION),
+					errmsg("column \"%s\" is in index used as replica identity",
+						   get_attname(RelationGetRelid(rel), attnum, false)));
+
+		/* Otherwise, just remove the attnotnull marking and do nothing else. */
+		drop_orphaned_notnull(RelationGetRelid(rel), attnum);
 	}
+	else
+	{
+		readyRels = NIL;
+		dropconstraint_internal(rel, conTup, DROP_RESTRICT, recurse, false,
+								false, &readyRels, lockmode);
 
-	readyRels = NIL;
-	dropconstraint_internal(rel, conTup, DROP_RESTRICT, recurse, false,
-							false, &readyRels, lockmode);
-
-	heap_freetuple(conTup);
+		heap_freetuple(conTup);
+	}
 
 	InvokeObjectPostAlterHook(RelationRelationId,
 							  RelationGetRelid(rel), attnum);
@@ -7796,6 +7812,33 @@ set_attnotnull(List **wqueue, Relation rel, AttrNumber attnum, bool recurse,
 	return retval;
 }
 
+/*
+ * In rare cases, a column can end up with an "orphaned" attnotnull marking
+ * with no corresponding pg_constraint row.  If the user then does ALTER TABLE
+ * DROP NOT NULL, this takes care of resetting that.
+ */
+static void
+drop_orphaned_notnull(Oid relationOid, AttrNumber attnum)
+{
+	Relation	attr_rel;
+	HeapTuple	tuple;
+	Form_pg_attribute attForm;
+
+	attr_rel = table_open(AttributeRelationId, RowExclusiveLock);
+
+	tuple = SearchSysCacheCopyAttNum(relationOid, attnum);
+	if (!HeapTupleIsValid(tuple))
+		elog(ERROR, "cache lookup failed for attribute %d of relation %u",
+			 attnum, relationOid);
+	attForm = (Form_pg_attribute) GETSTRUCT(tuple);
+
+	attForm->attnotnull = false;
+	CatalogTupleUpdate(attr_rel, &tuple->t_self, tuple);
+
+	table_close(attr_rel, RowExclusiveLock);
+}
+
+
 /*
  * ALTER TABLE ALTER COLUMN SET NOT NULL
  *
diff --git a/src/test/regress/expected/constraints.out b/src/test/regress/expected/constraints.out
index d9d8408e86..f0c166dc4f 100644
--- a/src/test/regress/expected/constraints.out
+++ b/src/test/regress/expected/constraints.out
@@ -866,6 +866,61 @@ select conname, contype, conkey from pg_constraint where conrelid = 'notnull_tbl
  foobar  | n       | {1}
 (1 row)
 
+DROP TABLE notnull_tbl1;
+-- make sure attnotnull is reset correctly when a PK is dropped indirectly
+CREATE TABLE notnull_tbl1 (c0 int, c1 int, PRIMARY KEY (c0, c1));
+ALTER TABLE  notnull_tbl1 DROP c1;
+\d+ notnull_tbl1
+                               Table "public.notnull_tbl1"
+ Column |  Type   | Collation | Nullable | Default | Storage | Stats target | Description 
+--------+---------+-----------+----------+---------+---------+--------------+-------------
+ c0     | integer |           | not null |         | plain   |              | 
+
+ALTER TABLE  notnull_tbl1 ALTER c0 DROP NOT NULL;
+\d+ notnull_tbl1
+                               Table "public.notnull_tbl1"
+ Column |  Type   | Collation | Nullable | Default | Storage | Stats target | Description 
+--------+---------+-----------+----------+---------+---------+--------------+-------------
+ c0     | integer |           |          |         | plain   |              | 
+
+DROP TABLE notnull_tbl1;
+-- again
+CREATE DOMAIN notnull_dom1 AS INTEGER;
+CREATE TABLE notnull_tbl1 (c0 notnull_dom1, c1 int, PRIMARY KEY (c0, c1));
+DROP DOMAIN notnull_dom1 CASCADE;
+NOTICE:  drop cascades to column c0 of table notnull_tbl1
+\d+ notnull_tbl1
+                               Table "public.notnull_tbl1"
+ Column |  Type   | Collation | Nullable | Default | Storage | Stats target | Description 
+--------+---------+-----------+----------+---------+---------+--------------+-------------
+ c1     | integer |           | not null |         | plain   |              | 
+
+ALTER TABLE  notnull_tbl1 ALTER c1 SET NOT NULL;
+\d+ notnull_tbl1
+                               Table "public.notnull_tbl1"
+ Column |  Type   | Collation | Nullable | Default | Storage | Stats target | Description 
+--------+---------+-----------+----------+---------+---------+--------------+-------------
+ c1     | integer |           | not null |         | plain   |              | 
+Not-null constraints:
+    "notnull_tbl1_c1_not_null" NOT NULL "c1"
+
+DROP TABLE notnull_tbl1;
+-- with a REPLICA IDENTITY column
+CREATE DOMAIN notnull_dom1 AS INTEGER;
+CREATE TABLE notnull_tbl1 (c0 notnull_dom1, c1 int UNIQUE, PRIMARY KEY (c0, c1));
+ALTER TABLE notnull_tbl1 REPLICA IDENTITY USING INDEX notnull_tbl1_c1_key;
+DROP DOMAIN notnull_dom1 CASCADE;
+NOTICE:  drop cascades to column c0 of table notnull_tbl1
+ALTER TABLE  notnull_tbl1 ALTER c1 DROP NOT NULL;
+ERROR:  column "c1" is in index used as replica identity
+DROP TABLE notnull_tbl1;
+-- with an identity generated column
+CREATE DOMAIN notnull_dom1 AS INTEGER;
+CREATE TABLE notnull_tbl1 (c0 notnull_dom1, c1 int generated by default as identity, PRIMARY KEY (c0, c1));
+DROP DOMAIN notnull_dom1 CASCADE;
+NOTICE:  drop cascades to column c0 of table notnull_tbl1
+ALTER TABLE  notnull_tbl1 ALTER c1 DROP NOT NULL;
+ERROR:  column "c1" of relation "notnull_tbl1" is an identity column
 DROP TABLE notnull_tbl1;
 -- nope
 CREATE TABLE notnull_tbl2 (a INTEGER CONSTRAINT blah NOT NULL, b INTEGER CONSTRAINT blah NOT NULL);
diff --git a/src/test/regress/sql/constraints.sql b/src/test/regress/sql/constraints.sql
index 87d685ae39..e946c29f18 100644
--- a/src/test/regress/sql/constraints.sql
+++ b/src/test/regress/sql/constraints.sql
@@ -599,6 +599,35 @@ ALTER TABLE notnull_tbl1 ADD CONSTRAINT foobar NOT NULL a;
 select conname, contype, conkey from pg_constraint where conrelid = 'notnull_tbl1'::regclass;
 DROP TABLE notnull_tbl1;
 
+-- make sure attnotnull is reset correctly when a PK is dropped indirectly
+CREATE TABLE notnull_tbl1 (c0 int, c1 int, PRIMARY KEY (c0, c1));
+ALTER TABLE  notnull_tbl1 DROP c1;
+\d+ notnull_tbl1
+ALTER TABLE  notnull_tbl1 ALTER c0 DROP NOT NULL;
+\d+ notnull_tbl1
+DROP TABLE notnull_tbl1;
+-- again
+CREATE DOMAIN notnull_dom1 AS INTEGER;
+CREATE TABLE notnull_tbl1 (c0 notnull_dom1, c1 int, PRIMARY KEY (c0, c1));
+DROP DOMAIN notnull_dom1 CASCADE;
+\d+ notnull_tbl1
+ALTER TABLE  notnull_tbl1 ALTER c1 SET NOT NULL;
+\d+ notnull_tbl1
+DROP TABLE notnull_tbl1;
+-- with a REPLICA IDENTITY column
+CREATE DOMAIN notnull_dom1 AS INTEGER;
+CREATE TABLE notnull_tbl1 (c0 notnull_dom1, c1 int UNIQUE, PRIMARY KEY (c0, c1));
+ALTER TABLE notnull_tbl1 REPLICA IDENTITY USING INDEX notnull_tbl1_c1_key;
+DROP DOMAIN notnull_dom1 CASCADE;
+ALTER TABLE  notnull_tbl1 ALTER c1 DROP NOT NULL;
+DROP TABLE notnull_tbl1;
+-- with an identity generated column
+CREATE DOMAIN notnull_dom1 AS INTEGER;
+CREATE TABLE notnull_tbl1 (c0 notnull_dom1, c1 int generated by default as identity, PRIMARY KEY (c0, c1));
+DROP DOMAIN notnull_dom1 CASCADE;
+ALTER TABLE  notnull_tbl1 ALTER c1 DROP NOT NULL;
+DROP TABLE notnull_tbl1;
+
 -- nope
 CREATE TABLE notnull_tbl2 (a INTEGER CONSTRAINT blah NOT NULL, b INTEGER CONSTRAINT blah NOT NULL);
 
-- 
2.39.2

From fe2c032c958b5855434232c9b699ef75a887e249 Mon Sep 17 00:00:00 2001
From: Alvaro Herrera <alvherre@alvh.no-ip.org>
Date: Wed, 10 Apr 2024 19:10:52 +0200
Subject: [PATCH v2 2/2] support this in pg_dump

---
 src/bin/pg_dump/pg_dump.c | 30 ++++++++++++++++++++++--------
 1 file changed, 22 insertions(+), 8 deletions(-)

diff --git a/src/bin/pg_dump/pg_dump.c b/src/bin/pg_dump/pg_dump.c
index c52e961b30..7dcbfaf8dc 100644
--- a/src/bin/pg_dump/pg_dump.c
+++ b/src/bin/pg_dump/pg_dump.c
@@ -8788,12 +8788,15 @@ getTableAttrs(Archive *fout, TableInfo *tblinfo, int numTables)
 						 "), E',\n    ') AS attfdwoptions,\n");
 
 	/*
-	 * Find out any NOT NULL markings for each column.  In 17 and up we have
-	 * to read pg_constraint, and keep track whether it's NO INHERIT; in older
-	 * versions we rely on pg_attribute.attnotnull.
+	 * Find out any NOT NULL markings for each column.  In 17 and up we
+	 * read pg_constraint to obtain the constraint name.  notnull_noinherit
+	 * is set according to the NO INHERIT property.  For versions prior to 17,
+	 * we store an empty string as the name when a constraint is marked as
+	 * attnotnull (this cues dumpTableSchema to print the NOT NULL clause
+	 * without a name); also, such cases are never NO INHERIT.
 	 *
-	 * We also track whether the constraint was defined directly in this table
-	 * or via an ancestor, for binary upgrade.
+	 * We track in notnull_inh whether the constraint was defined directly in
+	 * this table or via an ancestor, for binary upgrade.
 	 *
 	 * Lastly, we need to know if the PK for the table involves each column;
 	 * for columns that are there we need a NOT NULL marking even if there's
@@ -8801,13 +8804,24 @@ getTableAttrs(Archive *fout, TableInfo *tblinfo, int numTables)
 	 * NULLs after the data is loaded when the PK is created, later in the
 	 * dump; for this case we add throwaway constraints that are dropped once
 	 * the PK is created.
+	 *
+	 * Another complication arises from columns that have attnotnull set, but
+	 * for which no corresponding not-null nor PK constraint exists.  This can
+	 * happen if, for example, a primary key is dropped indirectly -- say,
+	 * because one of its columns is dropped.  This is an irregular condition,
+	 * so we don't work hard to preserve it, and instead act as though an
+	 * unnamed not-null constraint exists.
 	 */
 	if (fout->remoteVersion >= 170000)
 		appendPQExpBufferStr(q,
-							 "co.conname AS notnull_name,\n"
-							 "co.connoinherit AS notnull_noinherit,\n"
+							 "CASE WHEN co.conname IS NOT NULL THEN co.conname "
+							 "  WHEN a.attnotnull AND copk.conname IS NULL THEN '' ELSE NULL END AS notnull_name,\n"
+							 "CASE WHEN co.conname IS NOT NULL THEN co.connoinherit "
+							 "  WHEN a.attnotnull THEN false ELSE NULL END AS notnull_noinherit,\n"
 							 "copk.conname IS NOT NULL as notnull_is_pk,\n"
-							 "coalesce(NOT co.conislocal, true) AS notnull_inh,\n");
+							 "CASE WHEN co.conname IS NOT NULL THEN "
+							 "  coalesce(NOT co.conislocal, true) "
+							 "ELSE false END as notnull_inh,\n");
 	else
 		appendPQExpBufferStr(q,
 							 "CASE WHEN a.attnotnull THEN '' ELSE NULL END AS notnull_name,\n"
-- 
2.39.2

From ce8fff0ccb568a601ac9f765da7b7891066f522b Mon Sep 17 00:00:00 2001
From: Alvaro Herrera <alvherre@alvh.no-ip.org>
Date: Thu, 11 Apr 2024 12:29:34 +0200
Subject: [PATCH] Better handle indirect constraint drops

It is possible for certain cases to remove constraints without
maintaining the attnotnull in its correct state; for example if you drop
a column that's part of the primary key, and the remaining columns don't
have not-null constraints, then we should reset the attnotnull flags for
those other columns, but we don't.  We can handle those cases better by
doing the attnotnull reset in RemoveConstraintById() instead of in
dropconstraint_internal().

However, there are some cases where we must not do so.  For example if
those other columns are in replica identity indexes or are generated
identity columns, we must keep the flag, even though it results in the
catalog inconsistency that no constraint supports the flag.

Because of these exceptions, we also include a pg_dump hack to include a
not-null constraint when the attnotnull flag is set even if no
pg_constraint row exists.  This part is undesirable but necessary,
because failing to handle the case can result in unrestorable dumps.
---
 src/backend/catalog/pg_constraint.c       | 116 +++++++++++++++++++++-
 src/backend/commands/tablecmds.c          | 108 ++++++++++----------
 src/bin/pg_dump/pg_dump.c                 |  30 ++++--
 src/test/regress/expected/constraints.out |  73 ++++++++++++++
 src/test/regress/sql/constraints.sql      |  34 +++++++
 5 files changed, 301 insertions(+), 60 deletions(-)

diff --git a/src/backend/catalog/pg_constraint.c b/src/backend/catalog/pg_constraint.c
index 604280d322..c48acf00b0 100644
--- a/src/backend/catalog/pg_constraint.c
+++ b/src/backend/catalog/pg_constraint.c
@@ -19,6 +19,7 @@
 #include "access/htup_details.h"
 #include "access/sysattr.h"
 #include "access/table.h"
+#include "access/xact.h"
 #include "catalog/catalog.h"
 #include "catalog/dependency.h"
 #include "catalog/heap.h"
@@ -916,6 +917,8 @@ RemoveConstraintById(Oid conId)
 	Relation	conDesc;
 	HeapTuple	tup;
 	Form_pg_constraint con;
+	bool		dropping_pk = false;
+	List	   *unconstrained_cols = NIL;
 
 	conDesc = table_open(ConstraintRelationId, RowExclusiveLock);
 
@@ -940,7 +943,9 @@ RemoveConstraintById(Oid conId)
 		/*
 		 * We need to update the relchecks count if it is a check constraint
 		 * being dropped.  This update will force backends to rebuild relcache
-		 * entries when we commit.
+		 * entries when we commit.  For not-null and primary key constraints,
+		 * obtain the list of columns affected, so that we can reset their
+		 * attnotnull flags below.
 		 */
 		if (con->contype == CONSTRAINT_CHECK)
 		{
@@ -967,6 +972,36 @@ RemoveConstraintById(Oid conId)
 
 			table_close(pgrel, RowExclusiveLock);
 		}
+		else if (con->contype == CONSTRAINT_NOTNULL)
+		{
+			unconstrained_cols = list_make1_int(extractNotNullColumn(tup));
+		}
+		else if (con->contype == CONSTRAINT_PRIMARY)
+		{
+			Datum		adatum;
+			ArrayType  *arr;
+			int			numkeys;
+			bool		isNull;
+			int16	   *attnums;
+
+			dropping_pk = true;
+
+			adatum = heap_getattr(tup, Anum_pg_constraint_conkey,
+								  RelationGetDescr(conDesc), &isNull);
+			if (isNull)
+				elog(ERROR, "null conkey for constraint %u", con->oid);
+			arr = DatumGetArrayTypeP(adatum);	/* ensure not toasted */
+			numkeys = ARR_DIMS(arr)[0];
+			if (ARR_NDIM(arr) != 1 ||
+				numkeys < 0 ||
+				ARR_HASNULL(arr) ||
+				ARR_ELEMTYPE(arr) != INT2OID)
+				elog(ERROR, "conkey is not a 1-D smallint array");
+			attnums = (int16 *) ARR_DATA_PTR(arr);
+
+			for (int i = 0; i < numkeys; i++)
+				unconstrained_cols = lappend_int(unconstrained_cols, attnums[i]);
+		}
 
 		/* Keep lock on constraint's rel until end of xact */
 		table_close(rel, NoLock);
@@ -986,6 +1021,85 @@ RemoveConstraintById(Oid conId)
 	/* Fry the constraint itself */
 	CatalogTupleDelete(conDesc, &tup->t_self);
 
+	/*
+	 * If this was a NOT NULL or the primary key, the constrained columns must
+	 * have had pg_attribute.attnotnull set.  See if we need to reset it, and
+	 * do so.
+	 */
+	if (unconstrained_cols)
+	{
+		Relation	tablerel;
+		Relation	attrel;
+		Bitmapset  *pkcols;
+		ListCell   *lc;
+
+		/* Make the above deletion visible */
+		CommandCounterIncrement();
+
+		tablerel = table_open(con->conrelid, NoLock);	/* already have lock */
+		attrel = table_open(AttributeRelationId, RowExclusiveLock);
+
+		/*
+		 * We want to test columns for their presence in the primary key, but
+		 * only if we're not dropping it.
+		 */
+		pkcols = dropping_pk ? NULL :
+			RelationGetIndexAttrBitmap(tablerel,
+									   INDEX_ATTR_BITMAP_PRIMARY_KEY);
+
+		foreach(lc, unconstrained_cols)
+		{
+			AttrNumber	attnum = lfirst_int(lc);
+			HeapTuple	atttup;
+			HeapTuple	contup;
+			Bitmapset  *ircols;
+			Form_pg_attribute attForm;
+
+			/*
+			 * Obtain pg_attribute tuple and verify conditions on it.  We use
+			 * a copy we can scribble on.
+			 */
+			atttup = SearchSysCacheCopyAttNum(con->conrelid, attnum);
+			if (!HeapTupleIsValid(atttup))
+				elog(ERROR, "cache lookup failed for attribute %d of relation %u",
+					 attnum, con->conrelid);
+			attForm = (Form_pg_attribute) GETSTRUCT(atttup);
+
+			/*
+			 * Since the above deletion has been made visible, we can now
+			 * search for any remaining constraints on this column (or these
+			 * columns, in the case we're dropping a multicol primary key.)
+			 * Then, verify whether any further NOT NULL or primary key
+			 * exists, and reset attnotnull if none.
+			 */
+			contup = findNotNullConstraintAttnum(con->conrelid, attnum);
+			if (contup ||
+				bms_is_member(attnum - FirstLowInvalidHeapAttributeNumber,
+							  pkcols))
+				continue;
+
+			/*
+			 * ... but don't do it if the column is in the replica identity or
+			 * it's a generated column
+			 */
+			if (attForm->attidentity != '\0')
+				continue;
+			ircols = RelationGetIndexAttrBitmap(tablerel, INDEX_ATTR_BITMAP_IDENTITY_KEY); /* XXX bms_free */
+			if (bms_is_member(attnum - FirstLowInvalidHeapAttributeNumber, ircols))
+				continue;
+
+			/* Reset attnotnull */
+			if (attForm->attnotnull)
+			{
+				attForm->attnotnull = false;
+				CatalogTupleUpdate(attrel, &atttup->t_self, atttup);
+			}
+		}
+
+		table_close(attrel, RowExclusiveLock);
+		table_close(tablerel, NoLock);
+	}
+
 	/* Clean up */
 	ReleaseSysCache(tup);
 	table_close(conDesc, RowExclusiveLock);
diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c
index 8a98a0af48..aa9c11760f 100644
--- a/src/backend/commands/tablecmds.c
+++ b/src/backend/commands/tablecmds.c
@@ -149,6 +149,7 @@ typedef enum AlterTablePass
 	AT_PASS_ALTER_TYPE,			/* ALTER COLUMN TYPE */
 	AT_PASS_ADD_COL,			/* ADD COLUMN */
 	AT_PASS_SET_EXPRESSION,		/* ALTER SET EXPRESSION */
+	AT_PASS_OLD_COL_ATTRS,		/* re-install attnotnull */
 	AT_PASS_OLD_INDEX,			/* re-add existing indexes */
 	AT_PASS_OLD_CONSTR,			/* re-add existing constraints */
 	/* We could support a RENAME COLUMN pass here, but not currently used */
@@ -7678,17 +7679,23 @@ ATExecDropNotNull(Relation rel, const char *colName, bool recurse,
 	}
 
 	/*
-	 * Find the constraint that makes this column NOT NULL.
+	 * Find the constraint that makes this column NOT NULL, and drop it if we
+	 * see one.  dropconstraint_internal() will do necessary consistency
+	 * checking.  If there isn't one, there are two possibilities: either the
+	 * column is marked attnotnull because it's part of the primary key, and
+	 * then we just throw an appropriate error; or it's a leftover marking that
+	 * we can remove.  However, before doing the latter, to avoid breaking
+	 * consistency any further, prevent this if the column is part of the
+	 * replica identity.
 	 */
 	conTup = findNotNullConstraint(RelationGetRelid(rel), colName);
 	if (conTup == NULL)
 	{
 		Bitmapset  *pkcols;
+		Bitmapset  *ircols;
 
 		/*
-		 * There's no not-null constraint, so throw an error.  If the column
-		 * is in a primary key, we can throw a specific error.  Otherwise,
-		 * this is unexpected.
+		 * If the column is in a primary key, throw a specific error message.
 		 */
 		pkcols = RelationGetIndexAttrBitmap(rel, INDEX_ATTR_BITMAP_PRIMARY_KEY);
 		if (bms_is_member(attnum - FirstLowInvalidHeapAttributeNumber,
@@ -7697,16 +7704,27 @@ ATExecDropNotNull(Relation rel, const char *colName, bool recurse,
 					errcode(ERRCODE_INVALID_TABLE_DEFINITION),
 					errmsg("column \"%s\" is in a primary key", colName));
 
-		/* this shouldn't happen */
-		elog(ERROR, "could not find not-null constraint on column \"%s\", relation \"%s\"",
-			 colName, RelationGetRelationName(rel));
+		/* Also throw an error if the column is in the replica identity */
+		ircols = RelationGetIndexAttrBitmap(rel, INDEX_ATTR_BITMAP_IDENTITY_KEY);
+		if (bms_is_member(attnum - FirstLowInvalidHeapAttributeNumber, ircols))
+			ereport(ERROR,
+					errcode(ERRCODE_INVALID_TABLE_DEFINITION),
+					errmsg("column \"%s\" is in index used as replica identity",
+						   get_attname(RelationGetRelid(rel), attnum, false)));
+
+		/* Otherwise, just remove the attnotnull marking and do nothing else. */
+		attTup->attnotnull = false;
+		CatalogTupleUpdate(attr_rel, &tuple->t_self, tuple);
 	}
+	else
+	{
+		/* The normal case: we have a pg_constraint row, remove it */
+		readyRels = NIL;
+		dropconstraint_internal(rel, conTup, DROP_RESTRICT, recurse, false,
+								false, &readyRels, lockmode);
 
-	readyRels = NIL;
-	dropconstraint_internal(rel, conTup, DROP_RESTRICT, recurse, false,
-							false, &readyRels, lockmode);
-
-	heap_freetuple(conTup);
+		heap_freetuple(conTup);
+	}
 
 	InvokeObjectPostAlterHook(RelationRelationId,
 							  RelationGetRelid(rel), attnum);
@@ -12933,10 +12951,10 @@ dropconstraint_internal(Relation rel, HeapTuple constraintTup, DropBehavior beha
 	List	   *children;
 	ListCell   *child;
 	bool		is_no_inherit_constraint = false;
-	bool		dropping_pk = false;
 	char	   *constrName;
 	List	   *unconstrained_cols = NIL;
-	char	   *colname;
+	char	   *colname = NULL;
+	bool		dropping_pk = false;
 
 	if (list_member_oid(*readyRels, RelationGetRelid(rel)))
 		return InvalidObjectAddress;
@@ -12968,10 +12986,12 @@ dropconstraint_internal(Relation rel, HeapTuple constraintTup, DropBehavior beha
 	 */
 	if (con->contype == CONSTRAINT_NOTNULL)
 	{
-		AttrNumber	colnum = extractNotNullColumn(constraintTup);
+		AttrNumber	colnum;
 
-		if (colnum != InvalidAttrNumber)
-			unconstrained_cols = list_make1_int(colnum);
+		colnum = extractNotNullColumn(constraintTup);
+		unconstrained_cols = list_make1_int(colnum);
+		colname = NameStr(TupleDescAttr(RelationGetDescr(rel),
+										colnum - 1)->attname);
 	}
 	else if (con->contype == CONSTRAINT_PRIMARY)
 	{
@@ -13027,9 +13047,8 @@ dropconstraint_internal(Relation rel, HeapTuple constraintTup, DropBehavior beha
 	performDeletion(&conobj, behavior, 0);
 
 	/*
-	 * If this was a NOT NULL or the primary key, the constrained columns must
-	 * have had pg_attribute.attnotnull set.  See if we need to reset it, and
-	 * do so.
+	 * If this was a NOT NULL or the primary key, verify that we still have
+	 * constraints to support GENERATED AS IDENTITY or the replica identity.
 	 */
 	if (unconstrained_cols)
 	{
@@ -13038,7 +13057,7 @@ dropconstraint_internal(Relation rel, HeapTuple constraintTup, DropBehavior beha
 		Bitmapset  *ircols;
 		ListCell   *lc;
 
-		/* Make the above deletion visible */
+		/* Make implicit attnotnull changes visible */
 		CommandCounterIncrement();
 
 		attrel = table_open(AttributeRelationId, RowExclusiveLock);
@@ -13058,27 +13077,26 @@ dropconstraint_internal(Relation rel, HeapTuple constraintTup, DropBehavior beha
 			HeapTuple	atttup;
 			HeapTuple	contup;
 			Form_pg_attribute attForm;
+			char		attidentity;
 
 			/*
-			 * Obtain pg_attribute tuple and verify conditions on it.  We use
-			 * a copy we can scribble on.
+			 * Obtain pg_attribute tuple and verify conditions on it.
 			 */
-			atttup = SearchSysCacheCopyAttNum(RelationGetRelid(rel), attnum);
+			atttup = SearchSysCacheAttNum(RelationGetRelid(rel), attnum);
 			if (!HeapTupleIsValid(atttup))
 				elog(ERROR, "cache lookup failed for attribute %d of relation %u",
 					 attnum, RelationGetRelid(rel));
 			attForm = (Form_pg_attribute) GETSTRUCT(atttup);
+			attidentity = attForm->attidentity;
+			ReleaseSysCache(atttup);
 
 			/*
 			 * Since the above deletion has been made visible, we can now
 			 * search for any remaining constraints on this column (or these
 			 * columns, in the case we're dropping a multicol primary key.)
 			 * Then, verify whether any further NOT NULL or primary key
-			 * exists, and reset attnotnull if none.
-			 *
-			 * However, if this is a generated identity column, abort the
-			 * whole thing with a specific error message, because the
-			 * constraint is required in that case.
+			 * exists: if none and this is a generated identity column or the
+			 * replica identity, abort the whole thing.
 			 */
 			contup = findNotNullConstraintAttnum(RelationGetRelid(rel), attnum);
 			if (contup ||
@@ -13090,7 +13108,7 @@ dropconstraint_internal(Relation rel, HeapTuple constraintTup, DropBehavior beha
 			 * It's not valid to drop the not-null constraint for a GENERATED
 			 * AS IDENTITY column.
 			 */
-			if (attForm->attidentity)
+			if (attidentity != '\0')
 				ereport(ERROR,
 						errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
 						errmsg("column \"%s\" of relation \"%s\" is an identity column",
@@ -13107,13 +13125,6 @@ dropconstraint_internal(Relation rel, HeapTuple constraintTup, DropBehavior beha
 						errcode(ERRCODE_INVALID_TABLE_DEFINITION),
 						errmsg("column \"%s\" is in index used as replica identity",
 							   get_attname(RelationGetRelid(rel), lfirst_int(lc), false)));
-
-			/* Reset attnotnull */
-			if (attForm->attnotnull)
-			{
-				attForm->attnotnull = false;
-				CatalogTupleUpdate(attrel, &atttup->t_self, atttup);
-			}
 		}
 		table_close(attrel, RowExclusiveLock);
 	}
@@ -13152,13 +13163,6 @@ dropconstraint_internal(Relation rel, HeapTuple constraintTup, DropBehavior beha
 				 errmsg("cannot remove constraint from only the partitioned table when partitions exist"),
 				 errhint("Do not specify the ONLY keyword.")));
 
-	/* For not-null constraints we recurse by column name */
-	if (con->contype == CONSTRAINT_NOTNULL)
-		colname = NameStr(TupleDescAttr(RelationGetDescr(rel),
-										linitial_int(unconstrained_cols) - 1)->attname);
-	else
-		colname = NULL;			/* keep compiler quiet */
-
 	foreach(child, children)
 	{
 		Oid			childrelid = lfirst_oid(child);
@@ -13174,8 +13178,8 @@ dropconstraint_internal(Relation rel, HeapTuple constraintTup, DropBehavior beha
 		CheckTableNotInUse(childrel, "ALTER TABLE");
 
 		/*
-		 * We search for not-null constraint by column number, and other
-		 * constraints by name.
+		 * We search for not-null constraints by column name, and others by
+		 * constraint name.
 		 */
 		if (con->contype == CONSTRAINT_NOTNULL)
 		{
@@ -14656,12 +14660,14 @@ ATPostAlterTypeParse(Oid oldId, Oid oldRelId, Oid refRelId, char *cmd,
 				else if (cmd->subtype == AT_SetAttNotNull)
 				{
 					/*
-					 * The parser will create AT_AttSetNotNull subcommands for
-					 * columns of PRIMARY KEY indexes/constraints, but we need
-					 * not do anything with them here, because the columns'
-					 * NOT NULL marks will already have been propagated into
-					 * the new table definition.
+					 * When doing alter primary key column type, Removing old
+					 * PK constraint as sub-command will reset attnotnull to
+					 * false in RemoveConstraintById(). We must set attnotnull
+					 * back to true before doing re-add the index sub-command.
 					 */
+					cmd->subtype = AT_SetAttNotNull;
+					tab->subcmds[AT_PASS_OLD_COL_ATTRS] =
+						lappend(tab->subcmds[AT_PASS_OLD_COL_ATTRS], cmd);
 				}
 				else
 					elog(ERROR, "unexpected statement subtype: %d",
diff --git a/src/bin/pg_dump/pg_dump.c b/src/bin/pg_dump/pg_dump.c
index c52e961b30..7dcbfaf8dc 100644
--- a/src/bin/pg_dump/pg_dump.c
+++ b/src/bin/pg_dump/pg_dump.c
@@ -8788,12 +8788,15 @@ getTableAttrs(Archive *fout, TableInfo *tblinfo, int numTables)
 						 "), E',\n    ') AS attfdwoptions,\n");
 
 	/*
-	 * Find out any NOT NULL markings for each column.  In 17 and up we have
-	 * to read pg_constraint, and keep track whether it's NO INHERIT; in older
-	 * versions we rely on pg_attribute.attnotnull.
+	 * Find out any NOT NULL markings for each column.  In 17 and up we
+	 * read pg_constraint to obtain the constraint name.  notnull_noinherit
+	 * is set according to the NO INHERIT property.  For versions prior to 17,
+	 * we store an empty string as the name when a constraint is marked as
+	 * attnotnull (this cues dumpTableSchema to print the NOT NULL clause
+	 * without a name); also, such cases are never NO INHERIT.
 	 *
-	 * We also track whether the constraint was defined directly in this table
-	 * or via an ancestor, for binary upgrade.
+	 * We track in notnull_inh whether the constraint was defined directly in
+	 * this table or via an ancestor, for binary upgrade.
 	 *
 	 * Lastly, we need to know if the PK for the table involves each column;
 	 * for columns that are there we need a NOT NULL marking even if there's
@@ -8801,13 +8804,24 @@ getTableAttrs(Archive *fout, TableInfo *tblinfo, int numTables)
 	 * NULLs after the data is loaded when the PK is created, later in the
 	 * dump; for this case we add throwaway constraints that are dropped once
 	 * the PK is created.
+	 *
+	 * Another complication arises from columns that have attnotnull set, but
+	 * for which no corresponding not-null nor PK constraint exists.  This can
+	 * happen if, for example, a primary key is dropped indirectly -- say,
+	 * because one of its columns is dropped.  This is an irregular condition,
+	 * so we don't work hard to preserve it, and instead act as though an
+	 * unnamed not-null constraint exists.
 	 */
 	if (fout->remoteVersion >= 170000)
 		appendPQExpBufferStr(q,
-							 "co.conname AS notnull_name,\n"
-							 "co.connoinherit AS notnull_noinherit,\n"
+							 "CASE WHEN co.conname IS NOT NULL THEN co.conname "
+							 "  WHEN a.attnotnull AND copk.conname IS NULL THEN '' ELSE NULL END AS notnull_name,\n"
+							 "CASE WHEN co.conname IS NOT NULL THEN co.connoinherit "
+							 "  WHEN a.attnotnull THEN false ELSE NULL END AS notnull_noinherit,\n"
 							 "copk.conname IS NOT NULL as notnull_is_pk,\n"
-							 "coalesce(NOT co.conislocal, true) AS notnull_inh,\n");
+							 "CASE WHEN co.conname IS NOT NULL THEN "
+							 "  coalesce(NOT co.conislocal, true) "
+							 "ELSE false END as notnull_inh,\n");
 	else
 		appendPQExpBufferStr(q,
 							 "CASE WHEN a.attnotnull THEN '' ELSE NULL END AS notnull_name,\n"
diff --git a/src/test/regress/expected/constraints.out b/src/test/regress/expected/constraints.out
index d9d8408e86..34927f7afc 100644
--- a/src/test/regress/expected/constraints.out
+++ b/src/test/regress/expected/constraints.out
@@ -867,6 +867,79 @@ select conname, contype, conkey from pg_constraint where conrelid = 'notnull_tbl
 (1 row)
 
 DROP TABLE notnull_tbl1;
+-- make sure attnotnull is reset correctly when a PK is dropped indirectly,
+-- or kept if there's a reason for that
+CREATE TABLE notnull_tbl1 (c0 int, c1 int, PRIMARY KEY (c0, c1));
+ALTER TABLE  notnull_tbl1 DROP c1;
+\d+ notnull_tbl1
+                               Table "public.notnull_tbl1"
+ Column |  Type   | Collation | Nullable | Default | Storage | Stats target | Description 
+--------+---------+-----------+----------+---------+---------+--------------+-------------
+ c0     | integer |           |          |         | plain   |              | 
+
+DROP TABLE notnull_tbl1;
+-- same, via dropping a domain
+CREATE DOMAIN notnull_dom1 AS INTEGER;
+CREATE TABLE notnull_tbl1 (c0 notnull_dom1, c1 int, PRIMARY KEY (c0, c1));
+DROP DOMAIN notnull_dom1 CASCADE;
+NOTICE:  drop cascades to column c0 of table notnull_tbl1
+\d+ notnull_tbl1
+                               Table "public.notnull_tbl1"
+ Column |  Type   | Collation | Nullable | Default | Storage | Stats target | Description 
+--------+---------+-----------+----------+---------+---------+--------------+-------------
+ c1     | integer |           |          |         | plain   |              | 
+
+DROP TABLE notnull_tbl1;
+-- with a REPLICA IDENTITY column.  Here the not-nulls must be kept
+CREATE DOMAIN notnull_dom1 AS INTEGER;
+CREATE TABLE notnull_tbl1 (c0 notnull_dom1, c1 int UNIQUE, c2 int generated by default as identity, PRIMARY KEY (c0, c1, c2));
+ALTER TABLE notnull_tbl1 DROP CONSTRAINT notnull_tbl1_c2_not_null;
+ALTER TABLE notnull_tbl1 REPLICA IDENTITY USING INDEX notnull_tbl1_c1_key;
+DROP DOMAIN notnull_dom1 CASCADE;
+NOTICE:  drop cascades to column c0 of table notnull_tbl1
+ALTER TABLE  notnull_tbl1 ALTER c1 DROP NOT NULL;	-- can't be dropped
+ERROR:  column "c1" is in index used as replica identity
+ALTER TABLE  notnull_tbl1 ALTER c1 SET NOT NULL;	-- can be set right
+\d+ notnull_tbl1
+                                            Table "public.notnull_tbl1"
+ Column |  Type   | Collation | Nullable |             Default              | Storage | Stats target | Description 
+--------+---------+-----------+----------+----------------------------------+---------+--------------+-------------
+ c1     | integer |           | not null |                                  | plain   |              | 
+ c2     | integer |           | not null | generated by default as identity | plain   |              | 
+Indexes:
+    "notnull_tbl1_c1_key" UNIQUE CONSTRAINT, btree (c1) REPLICA IDENTITY
+Not-null constraints:
+    "notnull_tbl1_c1_not_null" NOT NULL "c1"
+
+-- Leave this table around for pg_upgrade testing
+CREATE DOMAIN notnull_dom2 AS INTEGER;
+CREATE TABLE notnull_tbl2 (c0 notnull_dom2, c1 int UNIQUE, c2 int generated by default as identity, PRIMARY KEY (c0, c1, c2));
+ALTER TABLE notnull_tbl2 DROP CONSTRAINT notnull_tbl2_c2_not_null;
+ALTER TABLE notnull_tbl2 REPLICA IDENTITY USING INDEX notnull_tbl2_c1_key;
+DROP DOMAIN notnull_dom2 CASCADE;
+NOTICE:  drop cascades to column c0 of table notnull_tbl2
+\d+ notnull_tbl2
+                                            Table "public.notnull_tbl2"
+ Column |  Type   | Collation | Nullable |             Default              | Storage | Stats target | Description 
+--------+---------+-----------+----------+----------------------------------+---------+--------------+-------------
+ c1     | integer |           | not null |                                  | plain   |              | 
+ c2     | integer |           | not null | generated by default as identity | plain   |              | 
+Indexes:
+    "notnull_tbl2_c1_key" UNIQUE CONSTRAINT, btree (c1) REPLICA IDENTITY
+
+ALTER TABLE notnull_tbl2 REPLICA IDENTITY FULL, ALTER c2 DROP IDENTITY;
+ALTER TABLE notnull_tbl2 ALTER c1 DROP NOT NULL, ALTER c2 DROP NOT NULL;
+\d+ notnull_tbl2
+                               Table "public.notnull_tbl2"
+ Column |  Type   | Collation | Nullable | Default | Storage | Stats target | Description 
+--------+---------+-----------+----------+---------+---------+--------------+-------------
+ c1     | integer |           |          |         | plain   |              | 
+ c2     | integer |           |          |         | plain   |              | 
+Indexes:
+    "notnull_tbl2_c1_key" UNIQUE CONSTRAINT, btree (c1)
+Replica Identity: FULL
+
+DROP TABLE notnull_tbl2;
 -- nope
 CREATE TABLE notnull_tbl2 (a INTEGER CONSTRAINT blah NOT NULL, b INTEGER CONSTRAINT blah NOT NULL);
 ERROR:  constraint "blah" for relation "notnull_tbl2" already exists
diff --git a/src/test/regress/sql/constraints.sql b/src/test/regress/sql/constraints.sql
index 87d685ae39..ecc7fb8dcf 100644
--- a/src/test/regress/sql/constraints.sql
+++ b/src/test/regress/sql/constraints.sql
@@ -599,6 +599,40 @@ ALTER TABLE notnull_tbl1 ADD CONSTRAINT foobar NOT NULL a;
 select conname, contype, conkey from pg_constraint where conrelid = 'notnull_tbl1'::regclass;
 DROP TABLE notnull_tbl1;
 
+-- make sure attnotnull is reset correctly when a PK is dropped indirectly,
+-- or kept if there's a reason for that
+CREATE TABLE notnull_tbl1 (c0 int, c1 int, PRIMARY KEY (c0, c1));
+ALTER TABLE  notnull_tbl1 DROP c1;
+\d+ notnull_tbl1
+DROP TABLE notnull_tbl1;
+-- same, via dropping a domain
+CREATE DOMAIN notnull_dom1 AS INTEGER;
+CREATE TABLE notnull_tbl1 (c0 notnull_dom1, c1 int, PRIMARY KEY (c0, c1));
+DROP DOMAIN notnull_dom1 CASCADE;
+\d+ notnull_tbl1
+DROP TABLE notnull_tbl1;
+-- with a REPLICA IDENTITY column.  Here the not-nulls must be kept
+CREATE DOMAIN notnull_dom1 AS INTEGER;
+CREATE TABLE notnull_tbl1 (c0 notnull_dom1, c1 int UNIQUE, c2 int generated by default as identity, PRIMARY KEY (c0, c1, c2));
+ALTER TABLE notnull_tbl1 DROP CONSTRAINT notnull_tbl1_c2_not_null;
+ALTER TABLE notnull_tbl1 REPLICA IDENTITY USING INDEX notnull_tbl1_c1_key;
+DROP DOMAIN notnull_dom1 CASCADE;
+ALTER TABLE  notnull_tbl1 ALTER c1 DROP NOT NULL;	-- can't be dropped
+ALTER TABLE  notnull_tbl1 ALTER c1 SET NOT NULL;	-- can be set right
+\d+ notnull_tbl1
+-- Leave this table around for pg_upgrade testing
+
+CREATE DOMAIN notnull_dom2 AS INTEGER;
+CREATE TABLE notnull_tbl2 (c0 notnull_dom2, c1 int UNIQUE, c2 int generated by default as identity, PRIMARY KEY (c0, c1, c2));
+ALTER TABLE notnull_tbl2 DROP CONSTRAINT notnull_tbl2_c2_not_null;
+ALTER TABLE notnull_tbl2 REPLICA IDENTITY USING INDEX notnull_tbl2_c1_key;
+DROP DOMAIN notnull_dom2 CASCADE;
+\d+ notnull_tbl2
+ALTER TABLE notnull_tbl2 REPLICA IDENTITY FULL, ALTER c2 DROP IDENTITY;
+ALTER TABLE notnull_tbl2 ALTER c1 DROP NOT NULL, ALTER c2 DROP NOT NULL;
+\d+ notnull_tbl2
+DROP TABLE notnull_tbl2;
+
 -- nope
 CREATE TABLE notnull_tbl2 (a INTEGER CONSTRAINT blah NOT NULL, b INTEGER CONSTRAINT blah NOT NULL);
 
-- 
2.39.2

From 28393502b3d11116fcf430264c10f931e948bedc Mon Sep 17 00:00:00 2001
From: Alvaro Herrera <alvherre@alvh.no-ip.org>
Date: Thu, 11 Apr 2024 12:29:34 +0200
Subject: [PATCH v2] Better handle indirect constraint drops

It is possible for certain cases to remove not-null constraints without
maintaining the attnotnull in its correct state; for example if you drop
a column that's part of the primary key, and the other columns of the PK don't
have not-null constraints, then we should reset the attnotnull flags for
those other columns; up to this commit, we didn't.  Handle those cases
better by doing the attnotnull reset in RemoveConstraintById() instead
of in dropconstraint_internal().

However, there are some cases where we must not do so.  For example if
those other columns are in replica identity indexes or are generated
identity columns, we must keep attnotnull set, even though it results in
the catalog inconsistency that no not-null constraint supports that.

Because the attnotnull reset now happens in more places than before, for
instance when a column of the primary key changes type, we need an
additional trick to reinstate it as necessary.  Introduce a new
alter-table pass that does this, which needs simply reschedule some
AT_SetAttNotNull subcommands that were already being generated and
ignored.

Because of the exceptions in which attnotnull is not reset noted above,
we also include a pg_dump hack to include a not-null constraint when the
attnotnull flag is set even if no pg_constraint row exists.  This part
is undesirable but necessary, because failing to handle the case can
result in unrestorable dumps.

Reported-by: Tender Wang <tndrwang@gmail.com>
Co-authored-by: Tender Wang <tndrwang@gmail.com>
Reviewed-by: jian he <jian.universality@gmail.com>
Discussion: https://postgr.es/m/CAHewXN=hMbNa3d43NOR=OCgdgpTt18S-1fmueCoEGesyeK4bqw@mail.gmail.com
---
 src/backend/catalog/pg_constraint.c       | 116 ++++++++++++++++++-
 src/backend/commands/tablecmds.c          | 135 +++++++++++-----------
 src/bin/pg_dump/pg_dump.c                 |  30 +++--
 src/test/regress/expected/constraints.out |  78 +++++++++++++
 src/test/regress/sql/constraints.sql      |  39 +++++++
 5 files changed, 322 insertions(+), 76 deletions(-)

diff --git a/src/backend/catalog/pg_constraint.c b/src/backend/catalog/pg_constraint.c
index 778b7c381d..56c083fe21 100644
--- a/src/backend/catalog/pg_constraint.c
+++ b/src/backend/catalog/pg_constraint.c
@@ -19,6 +19,7 @@
 #include "access/htup_details.h"
 #include "access/sysattr.h"
 #include "access/table.h"
+#include "access/xact.h"
 #include "catalog/catalog.h"
 #include "catalog/dependency.h"
 #include "catalog/heap.h"
@@ -933,6 +934,8 @@ RemoveConstraintById(Oid conId)
 	Relation	conDesc;
 	HeapTuple	tup;
 	Form_pg_constraint con;
+	bool		dropping_pk = false;
+	List	   *unconstrained_cols = NIL;
 
 	conDesc = table_open(ConstraintRelationId, RowExclusiveLock);
 
@@ -957,7 +960,9 @@ RemoveConstraintById(Oid conId)
 		/*
 		 * We need to update the relchecks count if it is a check constraint
 		 * being dropped.  This update will force backends to rebuild relcache
-		 * entries when we commit.
+		 * entries when we commit.  For not-null and primary key constraints,
+		 * obtain the list of columns affected, so that we can reset their
+		 * attnotnull flags below.
 		 */
 		if (con->contype == CONSTRAINT_CHECK)
 		{
@@ -984,6 +989,36 @@ RemoveConstraintById(Oid conId)
 
 			table_close(pgrel, RowExclusiveLock);
 		}
+		else if (con->contype == CONSTRAINT_NOTNULL)
+		{
+			unconstrained_cols = list_make1_int(extractNotNullColumn(tup));
+		}
+		else if (con->contype == CONSTRAINT_PRIMARY)
+		{
+			Datum		adatum;
+			ArrayType  *arr;
+			int			numkeys;
+			bool		isNull;
+			int16	   *attnums;
+
+			dropping_pk = true;
+
+			adatum = heap_getattr(tup, Anum_pg_constraint_conkey,
+								  RelationGetDescr(conDesc), &isNull);
+			if (isNull)
+				elog(ERROR, "null conkey for constraint %u", con->oid);
+			arr = DatumGetArrayTypeP(adatum);	/* ensure not toasted */
+			numkeys = ARR_DIMS(arr)[0];
+			if (ARR_NDIM(arr) != 1 ||
+				numkeys < 0 ||
+				ARR_HASNULL(arr) ||
+				ARR_ELEMTYPE(arr) != INT2OID)
+				elog(ERROR, "conkey is not a 1-D smallint array");
+			attnums = (int16 *) ARR_DATA_PTR(arr);
+
+			for (int i = 0; i < numkeys; i++)
+				unconstrained_cols = lappend_int(unconstrained_cols, attnums[i]);
+		}
 
 		/* Keep lock on constraint's rel until end of xact */
 		table_close(rel, NoLock);
@@ -1003,6 +1038,85 @@ RemoveConstraintById(Oid conId)
 	/* Fry the constraint itself */
 	CatalogTupleDelete(conDesc, &tup->t_self);
 
+	/*
+	 * If this was a NOT NULL or the primary key, the constrained columns must
+	 * have had pg_attribute.attnotnull set.  See if we need to reset it, and
+	 * do so.
+	 */
+	if (unconstrained_cols != NIL)
+	{
+		Relation	tablerel;
+		Relation	attrel;
+		Bitmapset  *pkcols;
+		ListCell   *lc;
+
+		/* Make the above deletion visible */
+		CommandCounterIncrement();
+
+		tablerel = table_open(con->conrelid, NoLock);	/* already have lock */
+		attrel = table_open(AttributeRelationId, RowExclusiveLock);
+
+		/*
+		 * We want to test columns for their presence in the primary key, but
+		 * only if we're not dropping it.
+		 */
+		pkcols = dropping_pk ? NULL :
+			RelationGetIndexAttrBitmap(tablerel,
+									   INDEX_ATTR_BITMAP_PRIMARY_KEY);
+
+		foreach(lc, unconstrained_cols)
+		{
+			AttrNumber	attnum = lfirst_int(lc);
+			HeapTuple	atttup;
+			HeapTuple	contup;
+			Bitmapset  *ircols;
+			Form_pg_attribute attForm;
+
+			/*
+			 * Obtain pg_attribute tuple and verify conditions on it.  We use
+			 * a copy we can scribble on.
+			 */
+			atttup = SearchSysCacheCopyAttNum(con->conrelid, attnum);
+			if (!HeapTupleIsValid(atttup))
+				elog(ERROR, "cache lookup failed for attribute %d of relation %u",
+					 attnum, con->conrelid);
+			attForm = (Form_pg_attribute) GETSTRUCT(atttup);
+
+			/*
+			 * Since the above deletion has been made visible, we can now
+			 * search for any remaining constraints on this column (or these
+			 * columns, in the case we're dropping a multicol primary key.)
+			 * Then, verify whether any further NOT NULL or primary key
+			 * exists, and reset attnotnull if none.
+			 */
+			contup = findNotNullConstraintAttnum(con->conrelid, attnum);
+			if (contup ||
+				bms_is_member(attnum - FirstLowInvalidHeapAttributeNumber,
+							  pkcols))
+				continue;
+
+			/*
+			 * ... but don't do it if the column is in the replica identity or
+			 * it's a generated column
+			 */
+			if (attForm->attidentity != '\0')
+				continue;
+			ircols = RelationGetIndexAttrBitmap(tablerel, INDEX_ATTR_BITMAP_IDENTITY_KEY);	/* XXX bms_free */
+			if (bms_is_member(attnum - FirstLowInvalidHeapAttributeNumber, ircols))
+				continue;
+
+			/* Reset attnotnull */
+			if (attForm->attnotnull)
+			{
+				attForm->attnotnull = false;
+				CatalogTupleUpdate(attrel, &atttup->t_self, atttup);
+			}
+		}
+
+		table_close(attrel, RowExclusiveLock);
+		table_close(tablerel, NoLock);
+	}
+
 	/* Clean up */
 	ReleaseSysCache(tup);
 	table_close(conDesc, RowExclusiveLock);
diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c
index f72b2dcadf..7b01c85f6b 100644
--- a/src/backend/commands/tablecmds.c
+++ b/src/backend/commands/tablecmds.c
@@ -149,6 +149,7 @@ typedef enum AlterTablePass
 	AT_PASS_ALTER_TYPE,			/* ALTER COLUMN TYPE */
 	AT_PASS_ADD_COL,			/* ADD COLUMN */
 	AT_PASS_SET_EXPRESSION,		/* ALTER SET EXPRESSION */
+	AT_PASS_OLD_COL_ATTRS,		/* re-install attnotnull */
 	AT_PASS_OLD_INDEX,			/* re-add existing indexes */
 	AT_PASS_OLD_CONSTR,			/* re-add existing constraints */
 	/* We could support a RENAME COLUMN pass here, but not currently used */
@@ -7662,17 +7663,23 @@ ATExecDropNotNull(Relation rel, const char *colName, bool recurse,
 	}
 
 	/*
-	 * Find the constraint that makes this column NOT NULL.
+	 * Find the constraint that makes this column NOT NULL, and drop it if we
+	 * see one.  dropconstraint_internal() will do necessary consistency
+	 * checking.  If there isn't one, there are two possibilities: either the
+	 * column is marked attnotnull because it's part of the primary key, and
+	 * then we just throw an appropriate error; or it's a leftover marking
+	 * that we can remove.  However, before doing the latter, to avoid
+	 * breaking consistency any further, prevent this if the column is part of
+	 * the replica identity.
 	 */
 	conTup = findNotNullConstraint(RelationGetRelid(rel), colName);
 	if (conTup == NULL)
 	{
 		Bitmapset  *pkcols;
+		Bitmapset  *ircols;
 
 		/*
-		 * There's no not-null constraint, so throw an error.  If the column
-		 * is in a primary key, we can throw a specific error.  Otherwise,
-		 * this is unexpected.
+		 * If the column is in a primary key, throw a specific error message.
 		 */
 		pkcols = RelationGetIndexAttrBitmap(rel, INDEX_ATTR_BITMAP_PRIMARY_KEY);
 		if (bms_is_member(attnum - FirstLowInvalidHeapAttributeNumber,
@@ -7681,16 +7688,27 @@ ATExecDropNotNull(Relation rel, const char *colName, bool recurse,
 					errcode(ERRCODE_INVALID_TABLE_DEFINITION),
 					errmsg("column \"%s\" is in a primary key", colName));
 
-		/* this shouldn't happen */
-		elog(ERROR, "could not find not-null constraint on column \"%s\", relation \"%s\"",
-			 colName, RelationGetRelationName(rel));
+		/* Also throw an error if the column is in the replica identity */
+		ircols = RelationGetIndexAttrBitmap(rel, INDEX_ATTR_BITMAP_IDENTITY_KEY);
+		if (bms_is_member(attnum - FirstLowInvalidHeapAttributeNumber, ircols))
+			ereport(ERROR,
+					errcode(ERRCODE_INVALID_TABLE_DEFINITION),
+					errmsg("column \"%s\" is in index used as replica identity",
+						   get_attname(RelationGetRelid(rel), attnum, false)));
+
+		/* Otherwise, just remove the attnotnull marking and do nothing else. */
+		attTup->attnotnull = false;
+		CatalogTupleUpdate(attr_rel, &tuple->t_self, tuple);
 	}
+	else
+	{
+		/* The normal case: we have a pg_constraint row, remove it */
+		readyRels = NIL;
+		dropconstraint_internal(rel, conTup, DROP_RESTRICT, recurse, false,
+								false, &readyRels, lockmode);
 
-	readyRels = NIL;
-	dropconstraint_internal(rel, conTup, DROP_RESTRICT, recurse, false,
-							false, &readyRels, lockmode);
-
-	heap_freetuple(conTup);
+		heap_freetuple(conTup);
+	}
 
 	InvokeObjectPostAlterHook(RelationRelationId,
 							  RelationGetRelid(rel), attnum);
@@ -12927,12 +12945,11 @@ dropconstraint_internal(Relation rel, HeapTuple constraintTup, DropBehavior beha
 	Form_pg_constraint con;
 	ObjectAddress conobj;
 	List	   *children;
-	ListCell   *child;
 	bool		is_no_inherit_constraint = false;
-	bool		dropping_pk = false;
 	char	   *constrName;
 	List	   *unconstrained_cols = NIL;
-	char	   *colname;
+	char	   *colname = NULL;
+	bool		dropping_pk = false;
 
 	if (list_member_oid(*readyRels, RelationGetRelid(rel)))
 		return InvalidObjectAddress;
@@ -12989,10 +13006,12 @@ dropconstraint_internal(Relation rel, HeapTuple constraintTup, DropBehavior beha
 	 */
 	if (con->contype == CONSTRAINT_NOTNULL)
 	{
-		AttrNumber	colnum = extractNotNullColumn(constraintTup);
+		AttrNumber	colnum;
 
-		if (colnum != InvalidAttrNumber)
-			unconstrained_cols = list_make1_int(colnum);
+		colnum = extractNotNullColumn(constraintTup);
+		unconstrained_cols = list_make1_int(colnum);
+		colname = NameStr(TupleDescAttr(RelationGetDescr(rel),
+										colnum - 1)->attname);
 	}
 	else if (con->contype == CONSTRAINT_PRIMARY)
 	{
@@ -13048,18 +13067,16 @@ dropconstraint_internal(Relation rel, HeapTuple constraintTup, DropBehavior beha
 	performDeletion(&conobj, behavior, 0);
 
 	/*
-	 * If this was a NOT NULL or the primary key, the constrained columns must
-	 * have had pg_attribute.attnotnull set.  See if we need to reset it, and
-	 * do so.
+	 * If this was a NOT NULL or the primary key, verify that we still have
+	 * constraints to support GENERATED AS IDENTITY or the replica identity.
 	 */
-	if (unconstrained_cols)
+	if (unconstrained_cols != NIL)
 	{
 		Relation	attrel;
 		Bitmapset  *pkcols;
 		Bitmapset  *ircols;
-		ListCell   *lc;
 
-		/* Make the above deletion visible */
+		/* Make implicit attnotnull changes visible */
 		CommandCounterIncrement();
 
 		attrel = table_open(AttributeRelationId, RowExclusiveLock);
@@ -13073,33 +13090,31 @@ dropconstraint_internal(Relation rel, HeapTuple constraintTup, DropBehavior beha
 									   INDEX_ATTR_BITMAP_PRIMARY_KEY);
 		ircols = RelationGetIndexAttrBitmap(rel, INDEX_ATTR_BITMAP_IDENTITY_KEY);
 
-		foreach(lc, unconstrained_cols)
+		foreach_int(attnum, unconstrained_cols)
 		{
-			AttrNumber	attnum = lfirst_int(lc);
 			HeapTuple	atttup;
 			HeapTuple	contup;
 			Form_pg_attribute attForm;
+			char		attidentity;
 
 			/*
-			 * Obtain pg_attribute tuple and verify conditions on it.  We use
-			 * a copy we can scribble on.
+			 * Obtain pg_attribute tuple and verify conditions on it.
 			 */
-			atttup = SearchSysCacheCopyAttNum(RelationGetRelid(rel), attnum);
+			atttup = SearchSysCacheAttNum(RelationGetRelid(rel), attnum);
 			if (!HeapTupleIsValid(atttup))
 				elog(ERROR, "cache lookup failed for attribute %d of relation %u",
 					 attnum, RelationGetRelid(rel));
 			attForm = (Form_pg_attribute) GETSTRUCT(atttup);
+			attidentity = attForm->attidentity;
+			ReleaseSysCache(atttup);
 
 			/*
 			 * Since the above deletion has been made visible, we can now
 			 * search for any remaining constraints on this column (or these
 			 * columns, in the case we're dropping a multicol primary key.)
 			 * Then, verify whether any further NOT NULL or primary key
-			 * exists, and reset attnotnull if none.
-			 *
-			 * However, if this is a generated identity column, abort the
-			 * whole thing with a specific error message, because the
-			 * constraint is required in that case.
+			 * exists: if none and this is a generated identity column or the
+			 * replica identity, abort the whole thing.
 			 */
 			contup = findNotNullConstraintAttnum(RelationGetRelid(rel), attnum);
 			if (contup ||
@@ -13111,7 +13126,7 @@ dropconstraint_internal(Relation rel, HeapTuple constraintTup, DropBehavior beha
 			 * It's not valid to drop the not-null constraint for a GENERATED
 			 * AS IDENTITY column.
 			 */
-			if (attForm->attidentity)
+			if (attidentity != '\0')
 				ereport(ERROR,
 						errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
 						errmsg("column \"%s\" of relation \"%s\" is an identity column",
@@ -13123,18 +13138,11 @@ dropconstraint_internal(Relation rel, HeapTuple constraintTup, DropBehavior beha
 			 * It's not valid to drop the not-null constraint for a column in
 			 * the replica identity index, either. (FULL is not affected.)
 			 */
-			if (bms_is_member(lfirst_int(lc) - FirstLowInvalidHeapAttributeNumber, ircols))
+			if (bms_is_member(attnum - FirstLowInvalidHeapAttributeNumber, ircols))
 				ereport(ERROR,
 						errcode(ERRCODE_INVALID_TABLE_DEFINITION),
 						errmsg("column \"%s\" is in index used as replica identity",
-							   get_attname(RelationGetRelid(rel), lfirst_int(lc), false)));
-
-			/* Reset attnotnull */
-			if (attForm->attnotnull)
-			{
-				attForm->attnotnull = false;
-				CatalogTupleUpdate(attrel, &atttup->t_self, atttup);
-			}
+							   get_attname(RelationGetRelid(rel), attnum, false)));
 		}
 		table_close(attrel, RowExclusiveLock);
 	}
@@ -13173,16 +13181,8 @@ dropconstraint_internal(Relation rel, HeapTuple constraintTup, DropBehavior beha
 				 errmsg("cannot remove constraint from only the partitioned table when partitions exist"),
 				 errhint("Do not specify the ONLY keyword.")));
 
-	/* For not-null constraints we recurse by column name */
-	if (con->contype == CONSTRAINT_NOTNULL)
-		colname = NameStr(TupleDescAttr(RelationGetDescr(rel),
-										linitial_int(unconstrained_cols) - 1)->attname);
-	else
-		colname = NULL;			/* keep compiler quiet */
-
-	foreach(child, children)
+	foreach_oid(childrelid, children)
 	{
-		Oid			childrelid = lfirst_oid(child);
 		Relation	childrel;
 		HeapTuple	tuple;
 		Form_pg_constraint childcon;
@@ -13195,8 +13195,8 @@ dropconstraint_internal(Relation rel, HeapTuple constraintTup, DropBehavior beha
 		CheckTableNotInUse(childrel, "ALTER TABLE");
 
 		/*
-		 * We search for not-null constraint by column number, and other
-		 * constraints by name.
+		 * We search for not-null constraints by column name, and others by
+		 * constraint name.
 		 */
 		if (con->contype == CONSTRAINT_NOTNULL)
 		{
@@ -13304,7 +13304,6 @@ dropconstraint_internal(Relation rel, HeapTuple constraintTup, DropBehavior beha
 		rel->rd_rel->relhassubclass)
 	{
 		List	   *colnames = NIL;
-		ListCell   *lc;
 		List	   *pkready = NIL;
 
 		/*
@@ -13317,15 +13316,14 @@ dropconstraint_internal(Relation rel, HeapTuple constraintTup, DropBehavior beha
 		 * Find out the list of column names to process.  Fortunately, we
 		 * already have the list of column numbers.
 		 */
-		foreach(lc, unconstrained_cols)
+		foreach_int(attnum, unconstrained_cols)
 		{
 			colnames = lappend(colnames, get_attname(RelationGetRelid(rel),
-													 lfirst_int(lc), false));
+													 attnum, false));
 		}
 
-		foreach(child, children)
+		foreach_oid(childrelid, children)
 		{
-			Oid			childrelid = lfirst_oid(child);
 			Relation	childrel;
 
 			if (list_member_oid(pkready, childrelid))
@@ -13335,10 +13333,9 @@ dropconstraint_internal(Relation rel, HeapTuple constraintTup, DropBehavior beha
 			childrel = table_open(childrelid, NoLock);
 			CheckTableNotInUse(childrel, "ALTER TABLE");
 
-			foreach(lc, colnames)
+			foreach_ptr(char, colName, colnames)
 			{
 				HeapTuple	contup;
-				char	   *colName = lfirst(lc);
 
 				contup = findNotNullConstraint(childrelid, colName);
 				if (contup == NULL)
@@ -14677,12 +14674,16 @@ ATPostAlterTypeParse(Oid oldId, Oid oldRelId, Oid refRelId, char *cmd,
 				else if (cmd->subtype == AT_SetAttNotNull)
 				{
 					/*
-					 * The parser will create AT_AttSetNotNull subcommands for
-					 * columns of PRIMARY KEY indexes/constraints, but we need
-					 * not do anything with them here, because the columns'
-					 * NOT NULL marks will already have been propagated into
-					 * the new table definition.
+					 * We see this subtype when a primary key is created
+					 * internally, for example when it is replaced with a new
+					 * constraint (say because one of the columns changes
+					 * type); in this case we need to reinstate attnotnull,
+					 * because it was removed because of the drop of the old
+					 * PK.  Schedule this subcommand to an upcoming AT pass.
 					 */
+					cmd->subtype = AT_SetAttNotNull;
+					tab->subcmds[AT_PASS_OLD_COL_ATTRS] =
+						lappend(tab->subcmds[AT_PASS_OLD_COL_ATTRS], cmd);
 				}
 				else
 					elog(ERROR, "unexpected statement subtype: %d",
diff --git a/src/bin/pg_dump/pg_dump.c b/src/bin/pg_dump/pg_dump.c
index 6d2f3fdef3..9edda90469 100644
--- a/src/bin/pg_dump/pg_dump.c
+++ b/src/bin/pg_dump/pg_dump.c
@@ -8788,12 +8788,15 @@ getTableAttrs(Archive *fout, TableInfo *tblinfo, int numTables)
 						 "), E',\n    ') AS attfdwoptions,\n");
 
 	/*
-	 * Find out any NOT NULL markings for each column.  In 17 and up we have
-	 * to read pg_constraint, and keep track whether it's NO INHERIT; in older
-	 * versions we rely on pg_attribute.attnotnull.
+	 * Find out any NOT NULL markings for each column.  In 17 and up we read
+	 * pg_constraint to obtain the constraint name.  notnull_noinherit is set
+	 * according to the NO INHERIT property.  For versions prior to 17, we
+	 * store an empty string as the name when a constraint is marked as
+	 * attnotnull (this cues dumpTableSchema to print the NOT NULL clause
+	 * without a name); also, such cases are never NO INHERIT.
 	 *
-	 * We also track whether the constraint was defined directly in this table
-	 * or via an ancestor, for binary upgrade.
+	 * We track in notnull_inh whether the constraint was defined directly in
+	 * this table or via an ancestor, for binary upgrade.
 	 *
 	 * Lastly, we need to know if the PK for the table involves each column;
 	 * for columns that are there we need a NOT NULL marking even if there's
@@ -8801,13 +8804,24 @@ getTableAttrs(Archive *fout, TableInfo *tblinfo, int numTables)
 	 * NULLs after the data is loaded when the PK is created, later in the
 	 * dump; for this case we add throwaway constraints that are dropped once
 	 * the PK is created.
+	 *
+	 * Another complication arises from columns that have attnotnull set, but
+	 * for which no corresponding not-null nor PK constraint exists.  This can
+	 * happen if, for example, a primary key is dropped indirectly -- say,
+	 * because one of its columns is dropped.  This is an irregular condition,
+	 * so we don't work hard to preserve it, and instead act as though an
+	 * unnamed not-null constraint exists.
 	 */
 	if (fout->remoteVersion >= 170000)
 		appendPQExpBufferStr(q,
-							 "co.conname AS notnull_name,\n"
-							 "co.connoinherit AS notnull_noinherit,\n"
+							 "CASE WHEN co.conname IS NOT NULL THEN co.conname "
+							 "  WHEN a.attnotnull AND copk.conname IS NULL THEN '' ELSE NULL END AS notnull_name,\n"
+							 "CASE WHEN co.conname IS NOT NULL THEN co.connoinherit "
+							 "  WHEN a.attnotnull THEN false ELSE NULL END AS notnull_noinherit,\n"
 							 "copk.conname IS NOT NULL as notnull_is_pk,\n"
-							 "coalesce(NOT co.conislocal, true) AS notnull_inh,\n");
+							 "CASE WHEN co.conname IS NOT NULL THEN "
+							 "  coalesce(NOT co.conislocal, true) "
+							 "ELSE false END as notnull_inh,\n");
 	else
 		appendPQExpBufferStr(q,
 							 "CASE WHEN a.attnotnull THEN '' ELSE NULL END AS notnull_name,\n"
diff --git a/src/test/regress/expected/constraints.out b/src/test/regress/expected/constraints.out
index d50dd1f61a..2fc0be7925 100644
--- a/src/test/regress/expected/constraints.out
+++ b/src/test/regress/expected/constraints.out
@@ -926,9 +926,87 @@ DROP TABLE notnull_tbl1;
 -- nope
 CREATE TABLE notnull_tbl2 (a INTEGER CONSTRAINT blah NOT NULL, b INTEGER CONSTRAINT blah NOT NULL);
 ERROR:  constraint "blah" for relation "notnull_tbl2" already exists
+-- can't drop not-null in primary key
 CREATE TABLE notnull_tbl2 (a INTEGER PRIMARY KEY);
 ALTER TABLE notnull_tbl2 ALTER a DROP NOT NULL;
 ERROR:  column "a" is in a primary key
+DROP TABLE notnull_tbl2;
+-- make sure attnotnull is reset correctly when a PK is dropped indirectly,
+-- or kept if there's a reason for that
+CREATE TABLE notnull_tbl1 (c0 int, c1 int, PRIMARY KEY (c0, c1));
+ALTER TABLE  notnull_tbl1 DROP c1;
+\d+ notnull_tbl1
+                               Table "public.notnull_tbl1"
+ Column |  Type   | Collation | Nullable | Default | Storage | Stats target | Description 
+--------+---------+-----------+----------+---------+---------+--------------+-------------
+ c0     | integer |           |          |         | plain   |              | 
+
+DROP TABLE notnull_tbl1;
+-- same, via dropping a domain
+CREATE DOMAIN notnull_dom1 AS INTEGER;
+CREATE TABLE notnull_tbl1 (c0 notnull_dom1, c1 int, PRIMARY KEY (c0, c1));
+DROP DOMAIN notnull_dom1 CASCADE;
+NOTICE:  drop cascades to column c0 of table notnull_tbl1
+\d+ notnull_tbl1
+                               Table "public.notnull_tbl1"
+ Column |  Type   | Collation | Nullable | Default | Storage | Stats target | Description 
+--------+---------+-----------+----------+---------+---------+--------------+-------------
+ c1     | integer |           |          |         | plain   |              | 
+
+DROP TABLE notnull_tbl1;
+-- with a REPLICA IDENTITY column.  Here the not-nulls must be kept
+CREATE DOMAIN notnull_dom1 AS INTEGER;
+CREATE TABLE notnull_tbl1 (c0 notnull_dom1, c1 int UNIQUE, c2 int generated by default as identity, PRIMARY KEY (c0, c1, c2));
+ALTER TABLE notnull_tbl1 DROP CONSTRAINT notnull_tbl1_c2_not_null;
+ALTER TABLE notnull_tbl1 REPLICA IDENTITY USING INDEX notnull_tbl1_c1_key;
+DROP DOMAIN notnull_dom1 CASCADE;
+NOTICE:  drop cascades to column c0 of table notnull_tbl1
+ALTER TABLE  notnull_tbl1 ALTER c1 DROP NOT NULL;	-- can't be dropped
+ERROR:  column "c1" is in index used as replica identity
+ALTER TABLE  notnull_tbl1 ALTER c1 SET NOT NULL;	-- can be set right
+\d+ notnull_tbl1
+                                            Table "public.notnull_tbl1"
+ Column |  Type   | Collation | Nullable |             Default              | Storage | Stats target | Description 
+--------+---------+-----------+----------+----------------------------------+---------+--------------+-------------
+ c1     | integer |           | not null |                                  | plain   |              | 
+ c2     | integer |           | not null | generated by default as identity | plain   |              | 
+Indexes:
+    "notnull_tbl1_c1_key" UNIQUE CONSTRAINT, btree (c1) REPLICA IDENTITY
+Not-null constraints:
+    "notnull_tbl1_c1_not_null" NOT NULL "c1"
+
+DROP TABLE notnull_tbl1;
+CREATE DOMAIN notnull_dom2 AS INTEGER;
+CREATE TABLE notnull_tbl2 (c0 notnull_dom2, c1 int UNIQUE, c2 int generated by default as identity, PRIMARY KEY (c0, c1, c2));
+ALTER TABLE notnull_tbl2 DROP CONSTRAINT notnull_tbl2_c2_not_null;
+ALTER TABLE notnull_tbl2 REPLICA IDENTITY USING INDEX notnull_tbl2_c1_key;
+DROP DOMAIN notnull_dom2 CASCADE;
+NOTICE:  drop cascades to column c0 of table notnull_tbl2
+\d+ notnull_tbl2
+                                            Table "public.notnull_tbl2"
+ Column |  Type   | Collation | Nullable |             Default              | Storage | Stats target | Description 
+--------+---------+-----------+----------+----------------------------------+---------+--------------+-------------
+ c1     | integer |           | not null |                                  | plain   |              | 
+ c2     | integer |           | not null | generated by default as identity | plain   |              | 
+Indexes:
+    "notnull_tbl2_c1_key" UNIQUE CONSTRAINT, btree (c1) REPLICA IDENTITY
+
+BEGIN;
+/* make sure the table can be put right, but roll that back */
+ALTER TABLE notnull_tbl2 REPLICA IDENTITY FULL, ALTER c2 DROP IDENTITY;
+ALTER TABLE notnull_tbl2 ALTER c1 DROP NOT NULL, ALTER c2 DROP NOT NULL;
+\d+ notnull_tbl2
+                               Table "public.notnull_tbl2"
+ Column |  Type   | Collation | Nullable | Default | Storage | Stats target | Description 
+--------+---------+-----------+----------+---------+---------+--------------+-------------
+ c1     | integer |           |          |         | plain   |              | 
+ c2     | integer |           |          |         | plain   |              | 
+Indexes:
+    "notnull_tbl2_c1_key" UNIQUE CONSTRAINT, btree (c1)
+Replica Identity: FULL
+
+ROLLBACK;
+-- Leave this table around for pg_upgrade testing
 CREATE TABLE notnull_tbl3 (a INTEGER NOT NULL, CHECK (a IS NOT NULL));
 ALTER TABLE notnull_tbl3 ALTER A DROP NOT NULL;
 ALTER TABLE notnull_tbl3 ADD b int, ADD CONSTRAINT pk PRIMARY KEY (a, b);
diff --git a/src/test/regress/sql/constraints.sql b/src/test/regress/sql/constraints.sql
index 7a39b504a3..8f85e72050 100644
--- a/src/test/regress/sql/constraints.sql
+++ b/src/test/regress/sql/constraints.sql
@@ -624,8 +624,47 @@ DROP TABLE notnull_tbl1;
 -- nope
 CREATE TABLE notnull_tbl2 (a INTEGER CONSTRAINT blah NOT NULL, b INTEGER CONSTRAINT blah NOT NULL);
 
+-- can't drop not-null in primary key
 CREATE TABLE notnull_tbl2 (a INTEGER PRIMARY KEY);
 ALTER TABLE notnull_tbl2 ALTER a DROP NOT NULL;
+DROP TABLE notnull_tbl2;
+
+-- make sure attnotnull is reset correctly when a PK is dropped indirectly,
+-- or kept if there's a reason for that
+CREATE TABLE notnull_tbl1 (c0 int, c1 int, PRIMARY KEY (c0, c1));
+ALTER TABLE  notnull_tbl1 DROP c1;
+\d+ notnull_tbl1
+DROP TABLE notnull_tbl1;
+-- same, via dropping a domain
+CREATE DOMAIN notnull_dom1 AS INTEGER;
+CREATE TABLE notnull_tbl1 (c0 notnull_dom1, c1 int, PRIMARY KEY (c0, c1));
+DROP DOMAIN notnull_dom1 CASCADE;
+\d+ notnull_tbl1
+DROP TABLE notnull_tbl1;
+-- with a REPLICA IDENTITY column.  Here the not-nulls must be kept
+CREATE DOMAIN notnull_dom1 AS INTEGER;
+CREATE TABLE notnull_tbl1 (c0 notnull_dom1, c1 int UNIQUE, c2 int generated by default as identity, PRIMARY KEY (c0, c1, c2));
+ALTER TABLE notnull_tbl1 DROP CONSTRAINT notnull_tbl1_c2_not_null;
+ALTER TABLE notnull_tbl1 REPLICA IDENTITY USING INDEX notnull_tbl1_c1_key;
+DROP DOMAIN notnull_dom1 CASCADE;
+ALTER TABLE  notnull_tbl1 ALTER c1 DROP NOT NULL;	-- can't be dropped
+ALTER TABLE  notnull_tbl1 ALTER c1 SET NOT NULL;	-- can be set right
+\d+ notnull_tbl1
+DROP TABLE notnull_tbl1;
+
+CREATE DOMAIN notnull_dom2 AS INTEGER;
+CREATE TABLE notnull_tbl2 (c0 notnull_dom2, c1 int UNIQUE, c2 int generated by default as identity, PRIMARY KEY (c0, c1, c2));
+ALTER TABLE notnull_tbl2 DROP CONSTRAINT notnull_tbl2_c2_not_null;
+ALTER TABLE notnull_tbl2 REPLICA IDENTITY USING INDEX notnull_tbl2_c1_key;
+DROP DOMAIN notnull_dom2 CASCADE;
+\d+ notnull_tbl2
+BEGIN;
+/* make sure the table can be put right, but roll that back */
+ALTER TABLE notnull_tbl2 REPLICA IDENTITY FULL, ALTER c2 DROP IDENTITY;
+ALTER TABLE notnull_tbl2 ALTER c1 DROP NOT NULL, ALTER c2 DROP NOT NULL;
+\d+ notnull_tbl2
+ROLLBACK;
+-- Leave this table around for pg_upgrade testing
 
 CREATE TABLE notnull_tbl3 (a INTEGER NOT NULL, CHECK (a IS NOT NULL));
 ALTER TABLE notnull_tbl3 ALTER A DROP NOT NULL;
-- 
2.39.2