Break transformCreateStmt() into multiple routines and make

transformAlterStmt() use these routines, instead of having lots of
duplicate (not to mention should-have-been-duplicate) code.
Adding a column with a CHECK constraint actually works now,
and the tests to reject unsupported DEFAULT and NOT NULL clauses
actually fire now.  ALTER TABLE ADD PRIMARY KEY works, modulo
having to have created the column(s) NOT NULL already.

1 files changed, 178 insertions, 187 deletions

diff --git a/src/backend/commands/command.c b/src/backend/commands/command.c
index 2b3cc08e2d4..a81d097d8b0 100644
--- a/src/backend/commands/command.c
+++ b/src/backend/commands/command.c
@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/commands/Attic/command.c,v 1.143 2001/10/05 17:28:11 tgl Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/commands/Attic/command.c,v 1.144 2001/10/12 00:07:14 tgl Exp $
  *
  * NOTES
  *	  The PerformAddAttribute() code, like most of the relation
@@ -309,9 +309,10 @@ AlterTableAddColumn(const char *relationName,
 	int			i;
 	int			minattnum,
 				maxatts;
-	Relation	idescs[Num_pg_attr_indices];
-	Relation	ridescs[Num_pg_class_indices];
-	bool		hasindex;
+	HeapTuple	typeTuple;
+	Form_pg_type tform;
+	char	   *typename;
+	int			attndims;
 
 	/*
 	 * permissions checking.  this would normally be done in utility.c,
@@ -339,57 +340,64 @@ AlterTableAddColumn(const char *relationName,
 	heap_close(rel, NoLock);	/* close rel but keep lock! */
 
 	/*
-	 * we can't add a not null attribute
-	 */
-	if (colDef->is_not_null)
-		elog(ERROR, "Can't add a NOT NULL attribute to an existing relation");
-
-	if (colDef->raw_default || colDef->cooked_default)
-		elog(ERROR, "Adding columns with defaults is not implemented.");
-
-
-	/*
-	 * if the first element in the 'schema' list is a "*" then we are
-	 * supposed to add this attribute to all classes that inherit from
-	 * 'relationName' (as well as to 'relationName').
+	 * Recurse to add the column to child classes, if requested.
 	 *
 	 * any permissions or problems with duplicate attributes will cause the
 	 * whole transaction to abort, which is what we want -- all or
 	 * nothing.
 	 */
-	if (colDef != NULL)
+	if (inherits)
 	{
-		if (inherits)
-		{
-			List	   *child,
-					   *children;
+		List	   *child,
+				   *children;
 
-			/* this routine is actually in the planner */
-			children = find_all_inheritors(myrelid);
+		/* this routine is actually in the planner */
+		children = find_all_inheritors(myrelid);
 
-			/*
-			 * find_all_inheritors does the recursive search of the
-			 * inheritance hierarchy, so all we have to do is process all
-			 * of the relids in the list that it returns.
-			 */
-			foreach(child, children)
-			{
-				Oid			childrelid = lfirsti(child);
-				char	   *childrelname;
+		/*
+		 * find_all_inheritors does the recursive search of the
+		 * inheritance hierarchy, so all we have to do is process all
+		 * of the relids in the list that it returns.
+		 */
+		foreach(child, children)
+		{
+			Oid			childrelid = lfirsti(child);
+			char	   *childrelname;
 
-				if (childrelid == myrelid)
-					continue;
-				rel = heap_open(childrelid, AccessExclusiveLock);
-				childrelname = pstrdup(RelationGetRelationName(rel));
-				heap_close(rel, AccessExclusiveLock);
+			if (childrelid == myrelid)
+				continue;
+			rel = heap_open(childrelid, AccessExclusiveLock);
+			childrelname = pstrdup(RelationGetRelationName(rel));
+			heap_close(rel, AccessExclusiveLock);
 
-				AlterTableAddColumn(childrelname, false, colDef);
+			AlterTableAddColumn(childrelname, false, colDef);
 
-				pfree(childrelname);
-			}
+			pfree(childrelname);
 		}
 	}
 
+	/*
+	 * OK, get on with it...
+	 *
+	 * Implementation restrictions: because we don't touch the table rows,
+	 * the new column values will initially appear to be NULLs.  (This
+	 * happens because the heap tuple access routines always check for
+	 * attnum > # of attributes in tuple, and return NULL if so.)  Therefore
+	 * we can't support a DEFAULT value in SQL92-compliant fashion, and
+	 * we also can't allow a NOT NULL constraint.
+	 *
+	 * We do allow CHECK constraints, even though these theoretically
+	 * could fail for NULL rows (eg, CHECK (newcol IS NOT NULL)).
+	 */
+	if (colDef->raw_default || colDef->cooked_default)
+		elog(ERROR, "Adding columns with defaults is not implemented."
+			 "\n\tAdd the column, then use ALTER TABLE SET DEFAULT.");
+
+	if (colDef->is_not_null)
+		elog(ERROR, "Adding NOT NULL columns is not implemented."
+			 "\n\tAdd the column, then use ALTER TABLE ADD CONSTRAINT.");
+
+
 	rel = heap_openr(RelationRelationName, RowExclusiveLock);
 
 	reltup = SearchSysCache(RELNAME,
@@ -400,22 +408,39 @@ AlterTableAddColumn(const char *relationName,
 		elog(ERROR, "ALTER TABLE: relation \"%s\" not found",
 			 relationName);
 
+	if (SearchSysCacheExists(ATTNAME,
+							 ObjectIdGetDatum(reltup->t_data->t_oid),
+							 PointerGetDatum(colDef->colname),
+							 0, 0))
+		elog(ERROR, "ALTER TABLE: column name \"%s\" already exists in table \"%s\"",
+			 colDef->colname, relationName);
+
 	minattnum = ((Form_pg_class) GETSTRUCT(reltup))->relnatts;
 	maxatts = minattnum + 1;
 	if (maxatts > MaxHeapAttributeNumber)
 		elog(ERROR, "ALTER TABLE: relations limited to %d columns",
 			 MaxHeapAttributeNumber);
+	i = minattnum + 1;
 
 	attrdesc = heap_openr(AttributeRelationName, RowExclusiveLock);
 
-	/*
-	 * Open all (if any) pg_attribute indices
-	 */
-	hasindex = RelationGetForm(attrdesc)->relhasindex;
-	if (hasindex)
-		CatalogOpenIndices(Num_pg_attr_indices, Name_pg_attr_indices, idescs);
+	if (colDef->typename->arrayBounds)
+	{
+		attndims = length(colDef->typename->arrayBounds);
+		typename = makeArrayTypeName(colDef->typename->name);
+	}
+	else
+	{
+		attndims = 0;
+		typename = colDef->typename->name;
+	}
 
-	attributeD.attrelid = reltup->t_data->t_oid;
+	typeTuple = SearchSysCache(TYPENAME,
+							   PointerGetDatum(typename),
+							   0, 0, 0);
+	if (!HeapTupleIsValid(typeTuple))
+		elog(ERROR, "ALTER TABLE: type \"%s\" does not exist", typename);
+	tform = (Form_pg_type) GETSTRUCT(typeTuple);
 
 	attributeTuple = heap_addheader(Natts_pg_attribute,
 									ATTRIBUTE_TUPLE_SIZE,
@@ -423,70 +448,36 @@ AlterTableAddColumn(const char *relationName,
 
 	attribute = (Form_pg_attribute) GETSTRUCT(attributeTuple);
 
-	i = 1 + minattnum;
-
+	attribute->attrelid = reltup->t_data->t_oid;
+	namestrcpy(&(attribute->attname), colDef->colname);
+	attribute->atttypid = typeTuple->t_data->t_oid;
+	attribute->attstattarget = DEFAULT_ATTSTATTARGET;
+	attribute->attlen = tform->typlen;
+	attribute->attcacheoff = -1;
+	attribute->atttypmod = colDef->typename->typmod;
+	attribute->attnum = i;
+	attribute->attbyval = tform->typbyval;
+	attribute->attndims = attndims;
+	attribute->attisset = (bool) (tform->typtype == 'c');
+	attribute->attstorage = tform->typstorage;
+	attribute->attalign = tform->typalign;
+	attribute->attnotnull = colDef->is_not_null;
+	attribute->atthasdef = (colDef->raw_default != NULL ||
+							colDef->cooked_default != NULL);
+
+	ReleaseSysCache(typeTuple);
+
+	heap_insert(attrdesc, attributeTuple);
+
+	/* Update indexes on pg_attribute */
+	if (RelationGetForm(attrdesc)->relhasindex)
 	{
-		HeapTuple	typeTuple;
-		Form_pg_type tform;
-		char	   *typename;
-		int			attndims;
-
-		if (SearchSysCacheExists(ATTNAME,
-								 ObjectIdGetDatum(reltup->t_data->t_oid),
-								 PointerGetDatum(colDef->colname),
-								 0, 0))
-			elog(ERROR, "ALTER TABLE: column name \"%s\" already exists in table \"%s\"",
-				 colDef->colname, relationName);
+		Relation	idescs[Num_pg_attr_indices];
 
-		/*
-		 * check to see if it is an array attribute.
-		 */
-
-		typename = colDef->typename->name;
-
-		if (colDef->typename->arrayBounds)
-		{
-			attndims = length(colDef->typename->arrayBounds);
-			typename = makeArrayTypeName(colDef->typename->name);
-		}
-		else
-			attndims = 0;
-
-		typeTuple = SearchSysCache(TYPENAME,
-								   PointerGetDatum(typename),
-								   0, 0, 0);
-		if (!HeapTupleIsValid(typeTuple))
-			elog(ERROR, "ALTER TABLE: type \"%s\" does not exist", typename);
-		tform = (Form_pg_type) GETSTRUCT(typeTuple);
-
-		namestrcpy(&(attribute->attname), colDef->colname);
-		attribute->atttypid = typeTuple->t_data->t_oid;
-		attribute->attlen = tform->typlen;
-		attribute->attstattarget = DEFAULT_ATTSTATTARGET;
-		attribute->attcacheoff = -1;
-		attribute->atttypmod = colDef->typename->typmod;
-		attribute->attnum = i;
-		attribute->attbyval = tform->typbyval;
-		attribute->attndims = attndims;
-		attribute->attisset = (bool) (tform->typtype == 'c');
-		attribute->attstorage = tform->typstorage;
-		attribute->attalign = tform->typalign;
-		attribute->attnotnull = false;
-		attribute->atthasdef = (colDef->raw_default != NULL ||
-								colDef->cooked_default != NULL);
-
-		ReleaseSysCache(typeTuple);
-
-		heap_insert(attrdesc, attributeTuple);
-		if (hasindex)
-			CatalogIndexInsert(idescs,
-							   Num_pg_attr_indices,
-							   attrdesc,
-							   attributeTuple);
-	}
-
-	if (hasindex)
+		CatalogOpenIndices(Num_pg_attr_indices, Name_pg_attr_indices, idescs);
+		CatalogIndexInsert(idescs, Num_pg_attr_indices, attrdesc, attributeTuple);
 		CatalogCloseIndices(Num_pg_attr_indices, idescs);
+	}
 
 	heap_close(attrdesc, NoLock);
 
@@ -499,9 +490,14 @@ AlterTableAddColumn(const char *relationName,
 	simple_heap_update(rel, &newreltup->t_self, newreltup);
 
 	/* keep catalog indices current */
-	CatalogOpenIndices(Num_pg_class_indices, Name_pg_class_indices, ridescs);
-	CatalogIndexInsert(ridescs, Num_pg_class_indices, rel, newreltup);
-	CatalogCloseIndices(Num_pg_class_indices, ridescs);
+	if (RelationGetForm(rel)->relhasindex)
+	{
+		Relation	ridescs[Num_pg_class_indices];
+
+		CatalogOpenIndices(Num_pg_class_indices, Name_pg_class_indices, ridescs);
+		CatalogIndexInsert(ridescs, Num_pg_class_indices, rel, newreltup);
+		CatalogCloseIndices(Num_pg_class_indices, ridescs);
+	}
 
 	heap_freetuple(newreltup);
 	ReleaseSysCache(reltup);
@@ -509,10 +505,27 @@ AlterTableAddColumn(const char *relationName,
 	heap_close(rel, NoLock);
 
 	/*
+	 * Make our catalog updates visible for subsequent steps.
+	 */
+	CommandCounterIncrement();
+
+	/*
+	 * Add any CHECK constraints attached to the new column.
+	 *
+	 * To do this we must re-open the rel so that its new attr list
+	 * gets loaded into the relcache.
+	 */
+	if (colDef->constraints != NIL)
+	{
+		rel = heap_openr(relationName, AccessExclusiveLock);
+		AddRelationRawConstraints(rel, NIL, colDef->constraints);
+		heap_close(rel, NoLock);
+	}
+
+	/*
 	 * Automatically create the secondary relation for TOAST if it
 	 * formerly had no such but now has toastable attributes.
 	 */
-	CommandCounterIncrement();
 	AlterTableCreateToastTable(relationName, true);
 }
 
@@ -596,7 +609,6 @@ AlterTableAlterColumnDefault(const char *relationName,
 	if (newDefault)
 	{
 		/* SET DEFAULT */
-		List	   *rawDefaults = NIL;
 		RawColumnDefault *rawEnt;
 
 		/* Get rid of the old one first */
@@ -605,13 +617,12 @@ AlterTableAlterColumnDefault(const char *relationName,
 		rawEnt = (RawColumnDefault *) palloc(sizeof(RawColumnDefault));
 		rawEnt->attnum = attnum;
 		rawEnt->raw_default = newDefault;
-		rawDefaults = lappend(rawDefaults, rawEnt);
 
 		/*
 		 * This function is intended for CREATE TABLE, so it processes a
 		 * _list_ of defaults, but we just do one.
 		 */
-		AddRelationRawConstraints(rel, rawDefaults, NIL);
+		AddRelationRawConstraints(rel, makeList1(rawEnt), NIL);
 	}
 	else
 	{
@@ -1169,10 +1180,11 @@ AlterTableDropColumn(const char *relationName,
  */
 void
 AlterTableAddConstraint(char *relationName,
-						bool inh, Node *newConstraint)
+						bool inh, List *newConstraints)
 {
-	if (newConstraint == NULL)
-		elog(ERROR, "ALTER TABLE / ADD CONSTRAINT passed invalid constraint.");
+	Relation	rel;
+	Oid	myrelid;
+	List	*listptr;
 
 #ifndef NO_SECURITY
 	if (!pg_ownercheck(GetUserId(), relationName, RELNAME))
@@ -1184,6 +1196,41 @@ AlterTableAddConstraint(char *relationName,
 		elog(ERROR, "ALTER TABLE ADD CONSTRAINT: %s is not a table",
 			 relationName);
 
+	rel = heap_openr(relationName, AccessExclusiveLock);
+	myrelid = RelationGetRelid(rel);
+
+	if (inh) {
+		List	   *child,
+				   *children;
+
+		/* this routine is actually in the planner */
+		children = find_all_inheritors(myrelid);
+
+		/*
+		 * find_all_inheritors does the recursive search of the
+		 * inheritance hierarchy, so all we have to do is process all
+		 * of the relids in the list that it returns.
+		 */
+		foreach(child, children)
+		{
+			Oid			childrelid = lfirsti(child);
+			char	   *childrelname;
+			Relation	childrel;
+
+			if (childrelid == myrelid)
+				continue;
+			childrel = heap_open(childrelid, AccessExclusiveLock);
+			childrelname = pstrdup(RelationGetRelationName(childrel));
+			heap_close(childrel, AccessExclusiveLock);
+			AlterTableAddConstraint(childrelname, false, newConstraints);
+			pfree(childrelname);
+		}
+	}
+
+	foreach(listptr, newConstraints)
+	{
+		Node   *newConstraint = lfirst(listptr);
+
 	switch (nodeTag(newConstraint))
 	{
 		case T_Constraint:
@@ -1202,33 +1249,14 @@ AlterTableAddConstraint(char *relationName,
 							HeapTuple	tuple;
 							RangeTblEntry *rte;
 							List	   *qual;
-							List	   *constlist;
-							Relation	rel;
 							Node	   *expr;
 							char	   *name;
-							Oid	myrelid;
 
 							if (constr->name)
 								name = constr->name;
 							else
 								name = "<unnamed>";
 
-							constlist = makeList1(constr);
-
-							rel = heap_openr(relationName, AccessExclusiveLock);
-							myrelid = RelationGetRelid(rel);
-
-							/* make sure it is not a view */
-							if (rel->rd_rel->relkind == RELKIND_VIEW)
-								elog(ERROR, "ALTER TABLE: cannot add constraint to a view");
-
-							/*
-							 * Scan all of the rows, looking for a false
-							 * match
-							 */
-							scan = heap_beginscan(rel, false, SnapshotNow, 0, NULL);
-							AssertState(scan != NULL);
-
 							/*
 							 * We need to make a parse state and range
 							 * table to allow us to transformExpr and
@@ -1280,6 +1308,8 @@ AlterTableAddConstraint(char *relationName,
 							 * Scan through the rows now, checking the
 							 * expression at each row.
 							 */
+							scan = heap_beginscan(rel, false, SnapshotNow, 0, NULL);
+
 							while (HeapTupleIsValid(tuple = heap_getnext(scan, 0)))
 							{
 								ExecStoreTuple(tuple, slot, InvalidBuffer, false);
@@ -1291,11 +1321,11 @@ AlterTableAddConstraint(char *relationName,
 								ResetExprContext(econtext);
 							}
 
+							heap_endscan(scan);
+
 							FreeExprContext(econtext);
 							pfree(slot);
 
-							heap_endscan(scan);
-
 							if (!successful)
 								elog(ERROR, "AlterTableAddConstraint: rejected due to CHECK constraint %s", name);
 
@@ -1306,38 +1336,8 @@ AlterTableAddConstraint(char *relationName,
 							 * the constraint against tuples already in
 							 * the table.
 							 */
-							AddRelationRawConstraints(rel, NIL, constlist);
-							heap_close(rel, NoLock);
-
-							if (inh) {
-								List	   *child,
-									   *children;
-
-								/* this routine is actually in the planner */
-								children = find_all_inheritors(myrelid);
-
-								/*
-								 * find_all_inheritors does the recursive search of the
-								 * inheritance hierarchy, so all we have to do is process all
-								 * of the relids in the list that it returns.
-								 */
-								foreach(child, children)
-								{
-									Oid			childrelid = lfirsti(child);
-									char	   *childrelname;
-	
-									if (childrelid == myrelid)
-										continue;
-									rel = heap_open(childrelid, AccessExclusiveLock);
-									childrelname = pstrdup(RelationGetRelationName(rel));
-									heap_close(rel, AccessExclusiveLock);
-
-									AlterTableAddConstraint(childrelname, false, newConstraint);
-	
-									pfree(childrelname);
-								}
-							}
-							pfree(constlist);
+							AddRelationRawConstraints(rel, NIL,
+													  makeList1(constr));
 
 							break;
 						}
@@ -1345,7 +1345,6 @@ AlterTableAddConstraint(char *relationName,
 						{
                             char  *iname = constr->name;
                             bool  istemp = is_temp_rel_name(relationName);
-                            Relation rel;
                             List	   *indexoidlist;
                             List     *indexoidscan;
                             Form_pg_attribute *rel_attrs;
@@ -1389,7 +1388,6 @@ AlterTableAddConstraint(char *relationName,
                             }
 
                             /* Need to check for unique key already on field(s) */
-                            rel = heap_openr(relationName, AccessExclusiveLock);
 
                             /*
                              * First we check for limited correctness of the
@@ -1490,9 +1488,6 @@ AlterTableAddConstraint(char *relationName,
                                elog(NOTICE, "Unique constraint supercedes existing index on relation \"%s\".  Drop the existing index to remove redundancy.", relationName);
                             pfree(iname);
 
-                            /* Finally, close relation */
-                            heap_close(rel, NoLock);
-
            					break;
                          }
 					default:
@@ -1503,8 +1498,7 @@ AlterTableAddConstraint(char *relationName,
 		case T_FkConstraint:
 			{
 				FkConstraint *fkconstraint = (FkConstraint *) newConstraint;
-				Relation	rel,
-							pkrel;
+				Relation	pkrel;
 				HeapScanDesc scan;
 				HeapTuple	tuple;
 				Trigger		trig;
@@ -1538,16 +1532,11 @@ AlterTableAddConstraint(char *relationName,
 						 fkconstraint->pktable_name);
 
 				/*
-				 * Grab an exclusive lock on the fk table, and then scan
-				 * through each tuple, calling the RI_FKey_Match_Ins
+				 * Scan through each tuple, calling the RI_FKey_Match_Ins
 				 * (insert trigger) as if that tuple had just been
 				 * inserted.  If any of those fail, it should elog(ERROR)
 				 * and that's that.
 				 */
-				rel = heap_openr(relationName, AccessExclusiveLock);
-				if (rel->rd_rel->relkind != RELKIND_RELATION)
-					elog(ERROR, "referencing table \"%s\" not a relation",
-						 relationName);
 
 				/*
 				 * First we check for limited correctness of the
@@ -1741,8 +1730,6 @@ AlterTableAddConstraint(char *relationName,
 					RI_FKey_check_ins(&fcinfo);
 				}
 				heap_endscan(scan);
-				heap_close(rel, NoLock);		/* close rel but keep
-												 * lock! */
 
 				pfree(trig.tgargs);
 				break;
@@ -1750,6 +1737,10 @@ AlterTableAddConstraint(char *relationName,
 		default:
 			elog(ERROR, "ALTER TABLE / ADD CONSTRAINT unable to determine type of constraint passed");
 	}
+	}
+
+	/* Close rel, but keep lock till commit */
+	heap_close(rel, NoLock);
 }