0001-Enable-CHECK-constraints-to-be-declared-NOT-VALID.patch

*** a/doc/src/sgml/catalogs.sgml
--- b/doc/src/sgml/catalogs.sgml
***************
*** 1898,1904 ****
        <entry><structfield>convalidated</structfield></entry>
        <entry><type>bool</type></entry>
        <entry></entry>
!       <entry>Has the constraint been validated? Can only be false for foreign keys</entry>
       </row>
  
       <row>
--- 1898,1904 ----
        <entry><structfield>convalidated</structfield></entry>
        <entry><type>bool</type></entry>
        <entry></entry>
!       <entry>Has the constraint been validated? Can only be false for foreign keys and CHECK constraints</entry>
       </row>
  
       <row>
*** a/doc/src/sgml/ref/alter_table.sgml
--- b/doc/src/sgml/ref/alter_table.sgml
***************
*** 240,246 **** ALTER TABLE <replaceable class="PARAMETER">name</replaceable>
      <listitem>
       <para>
        This form adds a new constraint to a table using the same syntax as
!       <xref linkend="SQL-CREATETABLE">. Newly added foreign key constraints can
        also be defined as <literal>NOT VALID</literal> to avoid the
        potentially lengthy initial check that must otherwise be performed.
        Constraint checks are skipped at create table time, so
--- 240,246 ----
      <listitem>
       <para>
        This form adds a new constraint to a table using the same syntax as
!       <xref linkend="SQL-CREATETABLE">. Newly added foreign key and CHECK constraints can
        also be defined as <literal>NOT VALID</literal> to avoid the
        potentially lengthy initial check that must otherwise be performed.
        Constraint checks are skipped at create table time, so
***************
*** 253,259 **** ALTER TABLE <replaceable class="PARAMETER">name</replaceable>
      <term><literal>VALIDATE CONSTRAINT</literal></term>
      <listitem>
       <para>
!       This form validates a foreign key constraint that was previously created
        as <literal>NOT VALID</literal>. Constraints already marked valid do not
        cause an error response.
       </para>
--- 253,259 ----
      <term><literal>VALIDATE CONSTRAINT</literal></term>
      <listitem>
       <para>
!       This form validates a foreign key or CHECK constraint that was previously created
        as <literal>NOT VALID</literal>. Constraints already marked valid do not
        cause an error response.
       </para>
*** a/src/backend/catalog/heap.c
--- b/src/backend/catalog/heap.c
***************
*** 97,103 **** static Oid AddNewRelationType(const char *typeName,
  				   Oid new_array_type);
  static void RelationRemoveInheritance(Oid relid);
  static void StoreRelCheck(Relation rel, char *ccname, Node *expr,
! 			  bool is_local, int inhcount);
  static void StoreConstraints(Relation rel, List *cooked_constraints);
  static bool MergeWithExistingConstraint(Relation rel, char *ccname, Node *expr,
  							bool allow_merge, bool is_local);
--- 97,103 ----
  				   Oid new_array_type);
  static void RelationRemoveInheritance(Oid relid);
  static void StoreRelCheck(Relation rel, char *ccname, Node *expr,
! 			  bool is_validated, bool is_local, int inhcount);
  static void StoreConstraints(Relation rel, List *cooked_constraints);
  static bool MergeWithExistingConstraint(Relation rel, char *ccname, Node *expr,
  							bool allow_merge, bool is_local);
***************
*** 1829,1835 **** StoreAttrDefault(Relation rel, AttrNumber attnum, Node *expr)
   */
  static void
  StoreRelCheck(Relation rel, char *ccname, Node *expr,
! 			  bool is_local, int inhcount)
  {
  	char	   *ccbin;
  	char	   *ccsrc;
--- 1829,1835 ----
   */
  static void
  StoreRelCheck(Relation rel, char *ccname, Node *expr,
! 			  bool is_validated, bool is_local, int inhcount)
  {
  	char	   *ccbin;
  	char	   *ccsrc;
***************
*** 1890,1896 **** StoreRelCheck(Relation rel, char *ccname, Node *expr,
  						  CONSTRAINT_CHECK,		/* Constraint Type */
  						  false,	/* Is Deferrable */
  						  false,	/* Is Deferred */
! 						  true, /* Is Validated */
  						  RelationGetRelid(rel),		/* relation */
  						  attNos,		/* attrs in the constraint */
  						  keycount,		/* # attrs in the constraint */
--- 1890,1896 ----
  						  CONSTRAINT_CHECK,		/* Constraint Type */
  						  false,	/* Is Deferrable */
  						  false,	/* Is Deferred */
! 						  is_validated,
  						  RelationGetRelid(rel),		/* relation */
  						  attNos,		/* attrs in the constraint */
  						  keycount,		/* # attrs in the constraint */
***************
*** 1950,1956 **** StoreConstraints(Relation rel, List *cooked_constraints)
  				StoreAttrDefault(rel, con->attnum, con->expr);
  				break;
  			case CONSTR_CHECK:
! 				StoreRelCheck(rel, con->name, con->expr,
  							  con->is_local, con->inhcount);
  				numchecks++;
  				break;
--- 1950,1956 ----
  				StoreAttrDefault(rel, con->attnum, con->expr);
  				break;
  			case CONSTR_CHECK:
! 				StoreRelCheck(rel, con->name, con->expr, !con->skip_validation,
  							  con->is_local, con->inhcount);
  				numchecks++;
  				break;
***************
*** 2064,2069 **** AddRelationNewConstraints(Relation rel,
--- 2064,2070 ----
  		cooked->name = NULL;
  		cooked->attnum = colDef->attnum;
  		cooked->expr = expr;
+ 		cooked->skip_validation = false;
  		cooked->is_local = is_local;
  		cooked->inhcount = is_local ? 0 : 1;
  		cookedConstraints = lappend(cookedConstraints, cooked);
***************
*** 2177,2183 **** AddRelationNewConstraints(Relation rel,
  		/*
  		 * OK, store it.
  		 */
! 		StoreRelCheck(rel, ccname, expr, is_local, is_local ? 0 : 1);
  
  		numchecks++;
  
--- 2178,2185 ----
  		/*
  		 * OK, store it.
  		 */
! 		StoreRelCheck(rel, ccname, expr, !cdef->skip_validation, is_local,
! 					  is_local ? 0 : 1);
  
  		numchecks++;
  
***************
*** 2186,2191 **** AddRelationNewConstraints(Relation rel,
--- 2188,2194 ----
  		cooked->name = ccname;
  		cooked->attnum = 0;
  		cooked->expr = expr;
+ 		cooked->skip_validation = cdef->skip_validation;
  		cooked->is_local = is_local;
  		cooked->inhcount = is_local ? 0 : 1;
  		cookedConstraints = lappend(cookedConstraints, cooked);
*** a/src/backend/commands/tablecmds.c
--- b/src/backend/commands/tablecmds.c
***************
*** 257,263 **** static void AlterIndexNamespaces(Relation classRel, Relation rel,
  static void AlterSeqNamespaces(Relation classRel, Relation rel,
  				   Oid oldNspOid, Oid newNspOid,
  				   const char *newNspName, LOCKMODE lockmode);
! static void ATExecValidateConstraint(Relation rel, const char *constrName);
  static int transformColumnNameList(Oid relId, List *colList,
  						int16 *attnums, Oid *atttypids);
  static int transformFkeyGetPrimaryKey(Relation pkrel, Oid *indexOid,
--- 257,264 ----
  static void AlterSeqNamespaces(Relation classRel, Relation rel,
  				   Oid oldNspOid, Oid newNspOid,
  				   const char *newNspName, LOCKMODE lockmode);
! static void ATExecValidateConstraint(Relation rel, const char *constrName,
! 						 bool recurse, bool recursing, LOCKMODE lockmode);
  static int transformColumnNameList(Oid relId, List *colList,
  						int16 *attnums, Oid *atttypids);
  static int transformFkeyGetPrimaryKey(Relation pkrel, Oid *indexOid,
***************
*** 268,273 **** static Oid transformFkeyCheckAttrs(Relation pkrel,
--- 269,276 ----
  						int numattrs, int16 *attnums,
  						Oid *opclasses);
  static void checkFkeyPermissions(Relation rel, int16 *attnums, int natts);
+ static void validateCheckConstraint(char *conname, Relation rel,
+ 						HeapTuple constrtup);
  static void validateForeignKeyConstraint(char *conname,
  							 Relation rel, Relation pkrel,
  							 Oid pkindOid, Oid constraintOid);
***************
*** 559,564 **** DefineRelation(CreateStmt *stmt, char relkind, Oid ownerId)
--- 562,568 ----
  			cooked->name = NULL;
  			cooked->attnum = attnum;
  			cooked->expr = colDef->cooked_default;
+ 			cooked->skip_validation = false;
  			cooked->is_local = true;	/* not used for defaults */
  			cooked->inhcount = 0;		/* ditto */
  			cookedDefaults = lappend(cookedDefaults, cooked);
***************
*** 1558,1563 **** MergeAttributes(List *schema, List *supers, char relpersistence,
--- 1562,1568 ----
  					cooked->name = pstrdup(name);
  					cooked->attnum = 0; /* not used for constraints */
  					cooked->expr = expr;
+ 					cooked->skip_validation = false;
  					cooked->is_local = false;
  					cooked->inhcount = 1;
  					constraints = lappend(constraints, cooked);
***************
*** 2922,2928 **** ATPrepCmd(List **wqueue, Relation rel, AlterTableCmd *cmd,
  			ATPrepAddInherit(rel);
  			pass = AT_PASS_MISC;
  			break;
! 		case AT_ValidateConstraint:
  		case AT_EnableTrig:		/* ENABLE TRIGGER variants */
  		case AT_EnableAlwaysTrig:
  		case AT_EnableReplicaTrig:
--- 2927,2940 ----
  			ATPrepAddInherit(rel);
  			pass = AT_PASS_MISC;
  			break;
! 		case AT_ValidateConstraint:		/* VALIDATE CONSTRAINT */
! 			ATSimplePermissions(rel, ATT_TABLE);
! 			/* Recursion occurs during execution phase */
! 			/* No command-specific prep needed except saving recurse flag */
! 			if (recurse)
! 				cmd->subtype = AT_ValidateConstraintRecurse;
! 			pass = AT_PASS_MISC;
! 			break;
  		case AT_EnableTrig:		/* ENABLE TRIGGER variants */
  		case AT_EnableAlwaysTrig:
  		case AT_EnableReplicaTrig:
***************
*** 3087,3094 **** ATExecCmd(List **wqueue, AlteredTableInfo *tab, Relation rel,
  		case AT_AddIndexConstraint:		/* ADD CONSTRAINT USING INDEX */
  			ATExecAddIndexConstraint(tab, rel, (IndexStmt *) cmd->def, lockmode);
  			break;
! 		case AT_ValidateConstraint:
! 			ATExecValidateConstraint(rel, cmd->name);
  			break;
  		case AT_DropConstraint:	/* DROP CONSTRAINT */
  			ATExecDropConstraint(rel, cmd->name, cmd->behavior,
--- 3099,3110 ----
  		case AT_AddIndexConstraint:		/* ADD CONSTRAINT USING INDEX */
  			ATExecAddIndexConstraint(tab, rel, (IndexStmt *) cmd->def, lockmode);
  			break;
! 		case AT_ValidateConstraint:		/* VALIDATE CONSTRAINT */
! 			ATExecValidateConstraint(rel, cmd->name, false, false, lockmode);
! 			break;
! 		case AT_ValidateConstraintRecurse:	/* VALIDATE CONSTRAINT with
! 											 * recursion */
! 			ATExecValidateConstraint(rel, cmd->name, true, false, lockmode);
  			break;
  		case AT_DropConstraint:	/* DROP CONSTRAINT */
  			ATExecDropConstraint(rel, cmd->name, cmd->behavior,
***************
*** 5361,5379 **** ATAddCheckConstraint(List **wqueue, AlteredTableInfo *tab, Relation rel,
  										list_make1(copyObject(constr)),
  										recursing, !recursing);
  
! 	/* Add each constraint to Phase 3's queue */
  	foreach(lcon, newcons)
  	{
  		CookedConstraint *ccon = (CookedConstraint *) lfirst(lcon);
- 		NewConstraint *newcon;
  
! 		newcon = (NewConstraint *) palloc0(sizeof(NewConstraint));
! 		newcon->name = ccon->name;
! 		newcon->contype = ccon->contype;
! 		/* ExecQual wants implicit-AND format */
! 		newcon->qual = (Node *) make_ands_implicit((Expr *) ccon->expr);
  
! 		tab->constraints = lappend(tab->constraints, newcon);
  
  		/* Save the actually assigned name if it was defaulted */
  		if (constr->conname == NULL)
--- 5377,5399 ----
  										list_make1(copyObject(constr)),
  										recursing, !recursing);
  
! 	/* Add each to-be-validated constraint to Phase 3's queue */
  	foreach(lcon, newcons)
  	{
  		CookedConstraint *ccon = (CookedConstraint *) lfirst(lcon);
  
! 		if (!ccon->skip_validation)
! 		{
! 			NewConstraint *newcon;
  
! 			newcon = (NewConstraint *) palloc0(sizeof(NewConstraint));
! 			newcon->name = ccon->name;
! 			newcon->contype = ccon->contype;
! 			/* ExecQual wants implicit-AND format */
! 			newcon->qual = (Node *) make_ands_implicit((Expr *) ccon->expr);
! 
! 			tab->constraints = lappend(tab->constraints, newcon);
! 		}
  
  		/* Save the actually assigned name if it was defaulted */
  		if (constr->conname == NULL)
***************
*** 5732,5740 **** ATAddForeignKeyConstraint(AlteredTableInfo *tab, Relation rel,
  
  /*
   * ALTER TABLE VALIDATE CONSTRAINT
   */
  static void
! ATExecValidateConstraint(Relation rel, const char *constrName)
  {
  	Relation	conrel;
  	SysScanDesc scan;
--- 5752,5766 ----
  
  /*
   * ALTER TABLE VALIDATE CONSTRAINT
+  *
+  * XXX The reason we handle recursion here rather than at Phase 1 is because
+  * there's no good way to skip recursing when handling foreign keys: there is
+  * no need to lock children in that case, yet we wouldn't be able to avoid
+  * doing so at that level.
   */
  static void
! ATExecValidateConstraint(Relation rel, const char *constrName, bool recurse,
! 						 bool recursing, LOCKMODE lockmode)
  {
  	Relation	conrel;
  	SysScanDesc scan;
***************
*** 5758,5765 **** ATExecValidateConstraint(Relation rel, const char *constrName)
  	while (HeapTupleIsValid(tuple = systable_getnext(scan)))
  	{
  		con = (Form_pg_constraint) GETSTRUCT(tuple);
! 		if (con->contype == CONSTRAINT_FOREIGN &&
! 			strcmp(NameStr(con->conname), constrName) == 0)
  		{
  			found = true;
  			break;
--- 5784,5790 ----
  	while (HeapTupleIsValid(tuple = systable_getnext(scan)))
  	{
  		con = (Form_pg_constraint) GETSTRUCT(tuple);
! 		if (strcmp(NameStr(con->conname), constrName) == 0)
  		{
  			found = true;
  			break;
***************
*** 5769,5807 **** ATExecValidateConstraint(Relation rel, const char *constrName)
  	if (!found)
  		ereport(ERROR,
  				(errcode(ERRCODE_UNDEFINED_OBJECT),
! 				 errmsg("foreign key constraint \"%s\" of relation \"%s\" does not exist",
  						constrName, RelationGetRelationName(rel))));
  
  	if (!con->convalidated)
  	{
! 		Oid			conid = HeapTupleGetOid(tuple);
! 		HeapTuple	copyTuple = heap_copytuple(tuple);
! 		Form_pg_constraint copy_con = (Form_pg_constraint) GETSTRUCT(copyTuple);
! 		Relation	refrel;
  
! 		/*
! 		 * Triggers are already in place on both tables, so a concurrent write
! 		 * that alters the result here is not possible. Normally we can run a
! 		 * query here to do the validation, which would only require
! 		 * AccessShareLock. In some cases, it is possible that we might need
! 		 * to fire triggers to perform the check, so we take a lock at
! 		 * RowShareLock level just in case.
! 		 */
! 		refrel = heap_open(con->confrelid, RowShareLock);
  
! 		validateForeignKeyConstraint((char *) constrName, rel, refrel,
! 									 con->conindid,
! 									 conid);
  
  		/*
  		 * Now update the catalog, while we have the door open.
  		 */
  		copy_con->convalidated = true;
  		simple_heap_update(conrel, &copyTuple->t_self, copyTuple);
  		CatalogUpdateIndexes(conrel, copyTuple);
  		heap_freetuple(copyTuple);
- 
- 		heap_close(refrel, NoLock);
  	}
  
  	systable_endscan(scan);
--- 5794,5897 ----
  	if (!found)
  		ereport(ERROR,
  				(errcode(ERRCODE_UNDEFINED_OBJECT),
! 				 errmsg("constraint \"%s\" of relation \"%s\" does not exist",
! 						constrName, RelationGetRelationName(rel))));
! 
! 	if (con->contype != CONSTRAINT_FOREIGN &&
! 		con->contype != CONSTRAINT_CHECK)
! 		ereport(ERROR,
! 				(errcode(ERRCODE_WRONG_OBJECT_TYPE),
! 				 errmsg("constraint \"%s\" of relation \"%s\" is not a foreign key or check constraint",
  						constrName, RelationGetRelationName(rel))));
  
  	if (!con->convalidated)
  	{
! 		HeapTuple	copyTuple;
! 		Form_pg_constraint copy_con;
  
! 		if (con->contype == CONSTRAINT_FOREIGN)
! 		{
! 			Oid			conid = HeapTupleGetOid(tuple);
! 			Relation	refrel;
  
! 			/*
! 			 * Triggers are already in place on both tables, so a concurrent write
! 			 * that alters the result here is not possible. Normally we can run a
! 			 * query here to do the validation, which would only require
! 			 * AccessShareLock. In some cases, it is possible that we might need
! 			 * to fire triggers to perform the check, so we take a lock at
! 			 * RowShareLock level just in case.
! 			 */
! 			refrel = heap_open(con->confrelid, RowShareLock);
! 
! 			validateForeignKeyConstraint((char *) constrName, rel, refrel,
! 										 con->conindid,
! 										 conid);
! 			heap_close(refrel, NoLock);
! 
! 			/*
! 			 * Foreign keys do not inherit, so we purposedly ignore the
! 			 * recursion bit here
! 			 */
! 		}
! 		else if (con->contype == CONSTRAINT_CHECK)
! 		{
! 			List	   *children = NIL;
! 			ListCell   *child;
! 
! 			/*
! 			 * If we're recursing, the parent has already done this, so skip
! 			 * it.
! 			 */
! 			if (!recursing)
! 				children = find_all_inheritors(RelationGetRelid(rel),
! 											   lockmode, NULL);
! 
! 			/*
! 			 * For CHECK constraints, we must ensure that we only mark the
! 			 * constraint as validated on the parent if it's already validated
! 			 * on the children.
! 			 *
! 			 * We recurse before validating on the parent, to reduce risk of
! 			 * deadlocks.
! 			 */
! 			foreach(child, children)
! 			{
! 				Oid childoid = lfirst_oid(child);
! 				Relation	childrel;
! 
! 				if (childoid == RelationGetRelid(rel))
! 					continue;
! 
! 				/*
! 				 * If we are told not to recurse, there had better not be any
! 				 * child tables; else the addition would put them out of step.
! 				 */
! 				if (!recurse)
! 					ereport(ERROR,
! 							(errcode(ERRCODE_INVALID_TABLE_DEFINITION),
! 							 errmsg("constraint must be validated on child tables too")));
! 
! 				/* find_all_inheritors already got lock */
! 				childrel = heap_open(childoid, NoLock);
! 
! 				ATExecValidateConstraint(childrel, constrName, false,
! 										 true, lockmode);
! 				heap_close(childrel, NoLock);
! 			}
! 
! 			validateCheckConstraint((char *) constrName, rel, tuple);
! 		}
  
  		/*
  		 * Now update the catalog, while we have the door open.
  		 */
+ 		copyTuple = heap_copytuple(tuple);
+ 		copy_con = (Form_pg_constraint) GETSTRUCT(copyTuple);
  		copy_con->convalidated = true;
  		simple_heap_update(conrel, &copyTuple->t_self, copyTuple);
  		CatalogUpdateIndexes(conrel, copyTuple);
  		heap_freetuple(copyTuple);
  	}
  
  	systable_endscan(scan);
***************
*** 6107,6112 **** checkFkeyPermissions(Relation rel, int16 *attnums, int natts)
--- 6197,6271 ----
  }
  
  /*
+  * Scan the existing rows in a table to verify they meet a proposed
+  * CHECK constraint.
+  *
+  * The caller must have opened and locked the relation appropriately.
+  */
+ static void
+ validateCheckConstraint(char *conname, Relation rel, HeapTuple constrtup)
+ {
+ 	EState		   *estate;
+ 	Datum			val;
+ 	char		   *conbin;
+ 	Expr		   *origexpr;
+ 	List		   *exprstate;
+ 	TupleDesc		tupdesc;
+ 	HeapScanDesc	scan;
+ 	HeapTuple		tuple;
+ 	ExprContext	   *econtext;
+ 	MemoryContext	oldcxt;
+ 	TupleTableSlot *slot;
+ 	bool			isnull;
+ 
+ 	estate = CreateExecutorState();
+ 	/*
+ 	 * XXX this tuple doesn't really come from a syscache, but this doesn't
+ 	 * matter to SysCacheGetAttr, because it only wants to be able to fetch the
+ 	 * tupdesc
+ 	 */
+ 	val = SysCacheGetAttr(CONSTROID, constrtup, Anum_pg_constraint_conbin,
+ 						  &isnull);
+ 	if (isnull)
+ 		elog(ERROR, "null conbin for constraint %u",
+ 			 HeapTupleGetOid(constrtup));
+ 	conbin = TextDatumGetCString(val);
+ 	origexpr = (Expr *) stringToNode(conbin);
+ 	exprstate = (List *) ExecPrepareExpr((Expr *) make_ands_implicit((Expr *) origexpr), estate);
+ 
+ 	econtext = GetPerTupleExprContext(estate);
+ 	tupdesc = RelationGetDescr(rel);
+ 	slot = MakeSingleTupleTableSlot(tupdesc);
+ 	econtext->ecxt_scantuple = slot;
+ 
+ 	scan = heap_beginscan(rel, SnapshotNow, 0, NULL);
+ 
+ 	/*
+ 	 * Switch to per-tuple memory context and reset it for each tuple
+ 	 * produced, so we don't leak memory.
+ 	 */
+ 	oldcxt = MemoryContextSwitchTo(GetPerTupleMemoryContext(estate));
+ 
+ 	while ((tuple = heap_getnext(scan, ForwardScanDirection)) != NULL)
+ 	{
+ 		ExecStoreTuple(tuple, slot, InvalidBuffer, false);
+ 
+ 		if (!ExecQual(exprstate, econtext, true))
+ 			ereport(ERROR,
+ 					(errcode(ERRCODE_CHECK_VIOLATION),
+ 					 errmsg("check constraint \"%s\" is violated by some row",
+ 							conname)));
+ 
+ 		ResetExprContext(econtext);
+ 	}
+ 
+ 	MemoryContextSwitchTo(oldcxt);
+ 	heap_endscan(scan);
+ 	ExecDropSingleTupleTableSlot(slot);
+ 	FreeExecutorState(estate);
+ }
+ 
+ /*
   * Scan the existing rows in a table to verify they meet a proposed FK
   * constraint.
   *
*** a/src/backend/parser/gram.y
--- b/src/backend/parser/gram.y
***************
*** 2746,2751 **** ConstraintElem:
--- 2746,2753 ----
  					n->location = @1;
  					n->raw_expr = $3;
  					n->cooked_expr = NULL;
+ 					n->skip_validation = false;
+ 					n->initially_valid = true;
  					if ($5 != 0)
  						ereport(ERROR,
  								(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
***************
*** 2753,2758 **** ConstraintElem:
--- 2755,2771 ----
  								 parser_errposition(@5)));
  					$$ = (Node *)n;
  				}
+ 			| CHECK '(' a_expr ')' NOT VALID
+ 				{
+ 					Constraint *n = makeNode(Constraint);
+ 					n->contype = CONSTR_CHECK;
+ 					n->location = @1;
+ 					n->raw_expr = $3;
+ 					n->cooked_expr = NULL;
+ 					n->skip_validation = true;
+ 					n->initially_valid = false;
+ 					$$ = (Node *)n;
+ 				}
  			| UNIQUE '(' columnList ')' opt_definition OptConsTableSpace
  				ConstraintAttributeSpec
  				{
*** a/src/include/catalog/heap.h
--- b/src/include/catalog/heap.h
***************
*** 30,35 **** typedef struct CookedConstraint
--- 30,36 ----
  	char	   *name;			/* name, or NULL if none */
  	AttrNumber	attnum;			/* which attr (only for DEFAULT) */
  	Node	   *expr;			/* transformed default or check expr */
+ 	bool		skip_validation;	/* skip validation? (only for CHECK) */
  	bool		is_local;		/* constraint has local (non-inherited) def */
  	int			inhcount;		/* number of times constraint is inherited */
  } CookedConstraint;
*** a/src/include/nodes/parsenodes.h
--- b/src/include/nodes/parsenodes.h
***************
*** 1190,1195 **** typedef enum AlterTableType
--- 1190,1196 ----
  	AT_AddConstraint,			/* add constraint */
  	AT_AddConstraintRecurse,	/* internal to commands/tablecmds.c */
  	AT_ValidateConstraint,		/* validate constraint */
+ 	AT_ValidateConstraintRecurse, /* internal to commands/tablecmds.c */
  	AT_ProcessedConstraint,		/* pre-processed add constraint (local in
  								 * parser/parse_utilcmd.c) */
  	AT_AddIndexConstraint,		/* add constraint using existing index */
***************
*** 1543,1548 **** typedef struct Constraint
--- 1544,1551 ----
  	char		fk_matchtype;	/* FULL, PARTIAL, UNSPECIFIED */
  	char		fk_upd_action;	/* ON UPDATE action */
  	char		fk_del_action;	/* ON DELETE action */
+ 
+ 	/* Fields used for constraints that allow a NOT VALID specification */
  	bool		skip_validation;	/* skip validation of existing rows? */
  	bool		initially_valid;	/* start the new constraint as valid */
  } Constraint;
*** a/src/test/regress/expected/alter_table.out
--- b/src/test/regress/expected/alter_table.out
***************
*** 196,205 **** DELETE FROM tmp3 where a=5;
--- 196,241 ----
  -- Try (and succeed) and repeat to show it works on already valid constraint
  ALTER TABLE tmp3 validate constraint tmpconstr;
  ALTER TABLE tmp3 validate constraint tmpconstr;
+ -- Try a non-verified CHECK constraint
+ ALTER TABLE tmp3 ADD CONSTRAINT b_greater_than_ten CHECK (b > 10); -- fail
+ ERROR:  check constraint "b_greater_than_ten" is violated by some row
+ ALTER TABLE tmp3 ADD CONSTRAINT b_greater_than_ten CHECK (b > 10) NOT VALID; -- succeeds
+ ALTER TABLE tmp3 VALIDATE CONSTRAINT b_greater_than_ten; -- fails
+ ERROR:  check constraint "b_greater_than_ten" is violated by some row
+ DELETE FROM tmp3 WHERE NOT b > 10;
+ ALTER TABLE tmp3 VALIDATE CONSTRAINT b_greater_than_ten; -- succeeds
+ ALTER TABLE tmp3 VALIDATE CONSTRAINT b_greater_than_ten; -- succeeds
+ -- Test inherited NOT VALID CHECK constraints
+ select * from tmp3;
+  a | b  
+ ---+----
+  1 | 20
+ (1 row)
+ 
+ CREATE TABLE tmp6 () INHERITS (tmp3);
+ CREATE TABLE tmp7 () INHERITS (tmp3);
+ INSERT INTO tmp6 VALUES (6, 30), (7, 16);
+ ALTER TABLE tmp3 ADD CONSTRAINT b_le_20 CHECK (b <= 20) NOT VALID;
+ ALTER TABLE tmp3 VALIDATE CONSTRAINT b_le_20;	-- fails
+ ERROR:  check constraint "b_le_20" is violated by some row
+ DELETE FROM tmp6 WHERE b > 20;
+ ALTER TABLE tmp3 VALIDATE CONSTRAINT b_le_20;	-- succeeds
+ -- An already validated constraint must not be revalidated
+ CREATE FUNCTION boo(int) RETURNS int IMMUTABLE STRICT LANGUAGE plpgsql AS $$ BEGIN RAISE NOTICE 'boo: %', $1; RETURN $1; END; $$;
+ INSERT INTO tmp7 VALUES (8, 18);
+ ALTER TABLE tmp7 ADD CONSTRAINT identity CHECK (b = boo(b));
+ NOTICE:  boo: 18
+ ALTER TABLE tmp3 ADD CONSTRAINT IDENTITY check (b = boo(b)) NOT VALID;
+ NOTICE:  merging constraint "identity" with inherited definition
+ ALTER TABLE tmp3 VALIDATE CONSTRAINT identity;
+ NOTICE:  boo: 16
+ NOTICE:  boo: 20
  -- Try (and fail) to create constraint from tmp5(a) to tmp4(a) - unique constraint on
  -- tmp4 is a,b
  ALTER TABLE tmp5 add constraint tmpconstr foreign key(a) references tmp4(a) match full;
  ERROR:  there is no unique constraint matching given keys for referenced table "tmp4"
+ DROP TABLE tmp7;
+ DROP TABLE tmp6;
  DROP TABLE tmp5;
  DROP TABLE tmp4;
  DROP TABLE tmp3;
*** a/src/test/regress/sql/alter_table.sql
--- b/src/test/regress/sql/alter_table.sql
***************
*** 236,247 **** DELETE FROM tmp3 where a=5;
--- 236,276 ----
  ALTER TABLE tmp3 validate constraint tmpconstr;
  ALTER TABLE tmp3 validate constraint tmpconstr;
  
+ -- Try a non-verified CHECK constraint
+ ALTER TABLE tmp3 ADD CONSTRAINT b_greater_than_ten CHECK (b > 10); -- fail
+ ALTER TABLE tmp3 ADD CONSTRAINT b_greater_than_ten CHECK (b > 10) NOT VALID; -- succeeds
+ ALTER TABLE tmp3 VALIDATE CONSTRAINT b_greater_than_ten; -- fails
+ DELETE FROM tmp3 WHERE NOT b > 10;
+ ALTER TABLE tmp3 VALIDATE CONSTRAINT b_greater_than_ten; -- succeeds
+ ALTER TABLE tmp3 VALIDATE CONSTRAINT b_greater_than_ten; -- succeeds
+ 
+ -- Test inherited NOT VALID CHECK constraints
+ select * from tmp3;
+ CREATE TABLE tmp6 () INHERITS (tmp3);
+ CREATE TABLE tmp7 () INHERITS (tmp3);
+ 
+ INSERT INTO tmp6 VALUES (6, 30), (7, 16);
+ ALTER TABLE tmp3 ADD CONSTRAINT b_le_20 CHECK (b <= 20) NOT VALID;
+ ALTER TABLE tmp3 VALIDATE CONSTRAINT b_le_20;	-- fails
+ DELETE FROM tmp6 WHERE b > 20;
+ ALTER TABLE tmp3 VALIDATE CONSTRAINT b_le_20;	-- succeeds
+ 
+ -- An already validated constraint must not be revalidated
+ CREATE FUNCTION boo(int) RETURNS int IMMUTABLE STRICT LANGUAGE plpgsql AS $$ BEGIN RAISE NOTICE 'boo: %', $1; RETURN $1; END; $$;
+ INSERT INTO tmp7 VALUES (8, 18);
+ ALTER TABLE tmp7 ADD CONSTRAINT identity CHECK (b = boo(b));
+ ALTER TABLE tmp3 ADD CONSTRAINT IDENTITY check (b = boo(b)) NOT VALID;
+ ALTER TABLE tmp3 VALIDATE CONSTRAINT identity;
  
  -- Try (and fail) to create constraint from tmp5(a) to tmp4(a) - unique constraint on
  -- tmp4 is a,b
  
  ALTER TABLE tmp5 add constraint tmpconstr foreign key(a) references tmp4(a) match full;
  
+ DROP TABLE tmp7;
+ 
+ DROP TABLE tmp6;
+ 
  DROP TABLE tmp5;
  
  DROP TABLE tmp4;