Make DROP TABLE rollback-able: postpone physical file delete until commit.

(WAL logging for this is not done yet, however.)  Clean up a number of really
crufty things that are no longer needed now that DROP behaves nicely.  Make
temp table mapper do the right things when drop or rename affecting a temp
table is rolled back.  Also, remove "relation modified while in use" error
check, in favor of locking tables at first reference and holding that lock
throughout the statement.

4 files changed, 114 insertions, 26 deletions

diff --git a/src/backend/parser/analyze.c b/src/backend/parser/analyze.c
index 93edc9f8781..63f39f2e4cb 100644
--- a/src/backend/parser/analyze.c
+++ b/src/backend/parser/analyze.c
@@ -6,7 +6,7 @@
  * Portions Copyright (c) 1996-2000, PostgreSQL, Inc
  * Portions Copyright (c) 1994, Regents of the University of California
  *
- *	$Id: analyze.c,v 1.164 2000/11/05 01:42:07 tgl Exp $
+ *	$Id: analyze.c,v 1.165 2000/11/08 22:09:58 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -115,7 +115,7 @@ static void
 release_pstate_resources(ParseState *pstate)
 {
 	if (pstate->p_target_relation != NULL)
-		heap_close(pstate->p_target_relation, AccessShareLock);
+		heap_close(pstate->p_target_relation, NoLock);
 	pstate->p_target_relation = NULL;
 	pstate->p_target_rangetblentry = NULL;
 }
@@ -292,6 +292,7 @@ transformDeleteStmt(ParseState *pstate, DeleteStmt *stmt)
 	qry->commandType = CMD_DELETE;
 
 	/* set up a range table */
+	lockTargetTable(pstate, stmt->relname);
 	makeRangeTable(pstate, NIL);
 	setTargetTable(pstate, stmt->relname, stmt->inh, true);
 
@@ -332,6 +333,13 @@ transformInsertStmt(ParseState *pstate, InsertStmt *stmt)
 	pstate->p_is_insert = true;
 
 	/*
+	 * Must get write lock on target table before scanning SELECT,
+	 * else we will grab the wrong kind of initial lock if the target
+	 * table is also mentioned in the SELECT part.
+	 */
+	lockTargetTable(pstate, stmt->relname);
+
+	/*
 	 * Is it INSERT ... SELECT or INSERT ... VALUES?
 	 */
 	if (stmt->selectStmt)
@@ -1522,6 +1530,16 @@ transformRuleStmt(ParseState *pstate, RuleStmt *stmt)
 	qry->utilityStmt = (Node *) stmt;
 
 	/*
+	 * To avoid deadlock, make sure the first thing we do is grab
+	 * AccessExclusiveLock on the target relation.  This will be
+	 * needed by DefineQueryRewrite(), and we don't want to grab a lesser
+	 * lock beforehand.  We don't need to hold a refcount on the relcache
+	 * entry, however.
+	 */
+	heap_close(heap_openr(stmt->object->relname, AccessExclusiveLock),
+			   NoLock);
+
+	/*
 	 * NOTE: 'OLD' must always have a varno equal to 1 and 'NEW'
 	 * equal to 2.  Set up their RTEs in the main pstate for use
 	 * in parsing the rule qualification.
@@ -1727,6 +1745,9 @@ transformSelectStmt(ParseState *pstate, SelectStmt *stmt)
 		qry->isBinary = FALSE;
 	}
 
+	/* make FOR UPDATE clause available to addRangeTableEntry */
+	pstate->p_forUpdate = stmt->forUpdate;
+
 	/* set up a range table */
 	makeRangeTable(pstate, stmt->fromClause);
 
@@ -1765,7 +1786,7 @@ transformSelectStmt(ParseState *pstate, SelectStmt *stmt)
 	qry->rtable = pstate->p_rtable;
 	qry->jointree = makeFromExpr(pstate->p_joinlist, qual);
 
-	if (stmt->forUpdate != NULL)
+	if (stmt->forUpdate != NIL)
 		transformForUpdate(qry, stmt->forUpdate);
 
 	return qry;
@@ -1951,7 +1972,7 @@ transformSetOperationStmt(ParseState *pstate, SelectStmt *stmt)
 	qry->rtable = pstate->p_rtable;
 	qry->jointree = makeFromExpr(pstate->p_joinlist, NULL);
 
-	if (forUpdate != NULL)
+	if (forUpdate != NIL)
 		transformForUpdate(qry, forUpdate);
 
 	return qry;
@@ -2159,6 +2180,7 @@ transformUpdateStmt(ParseState *pstate, UpdateStmt *stmt)
 	 * used in FROM, we'd fail to notice that it should be marked
 	 * checkForRead as well as checkForWrite.  See setTargetTable().
 	 */
+	lockTargetTable(pstate, stmt->relname);
 	makeRangeTable(pstate, stmt->fromClause);
 	setTargetTable(pstate, stmt->relname, stmt->inh, true);
 
diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y
index 4fbc628c582..5572828d259 100644
--- a/src/backend/parser/gram.y
+++ b/src/backend/parser/gram.y
@@ -11,7 +11,7 @@
  *
  *
  * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/parser/gram.y,v 2.207 2000/11/08 21:28:06 momjian Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/parser/gram.y,v 2.208 2000/11/08 22:09:58 tgl Exp $
  *
  * HISTORY
  *	  AUTHOR			DATE			MAJOR EVENT
@@ -334,7 +334,7 @@ static void doNegateFloat(Value *v);
  * when some sort of pg_privileges relation is introduced.
  * - Todd A. Brandys 1998-01-01?
  */
-%token	ABORT_TRANS, ACCESS, AFTER, AGGREGATE, ANALYZE, ANALYSE /* British */
+%token	ABORT_TRANS, ACCESS, AFTER, AGGREGATE, ANALYZE, ANALYSE,
 		BACKWARD, BEFORE, BINARY, BIT,
 		CACHE, CHECKPOINT, CLUSTER, COMMENT, COPY, CREATEDB, CREATEUSER, CYCLE,
 		DATABASE, DELIMITERS, DO,
@@ -2466,11 +2466,7 @@ ExtendStmt:  EXTEND INDEX index_name where_clause
 /* NOT USED
 RecipeStmt:  EXECUTE RECIPE recipe_name
 				{
-					RecipeStmt *n;
-					if (!IsTransactionBlock())
-						elog(ERROR,"EXECUTE RECIPE may only be used in begin/end transaction blocks");
-
-					n = makeNode(RecipeStmt);
+					RecipeStmt *n = makeNode(RecipeStmt);
 					n->recipeName = $3;
 					$$ = (Node *)n;
 				}
@@ -2633,8 +2629,6 @@ oper_argtypes:	Typename
 ReindexStmt:  REINDEX reindex_type name opt_force
 				{
 					ReindexStmt *n = makeNode(ReindexStmt);
-					if (IsTransactionBlock())
-						elog(ERROR,"REINDEX command could only be used outside begin/end transaction blocks");
 					n->reindexType = $2;
 					n->name = $3;
 					n->force = $4;
diff --git a/src/backend/parser/parse_clause.c b/src/backend/parser/parse_clause.c
index 38dc3ea0976..60521d13475 100644
--- a/src/backend/parser/parse_clause.c
+++ b/src/backend/parser/parse_clause.c
@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/parser/parse_clause.c,v 1.70 2000/10/07 00:58:17 tgl Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/parser/parse_clause.c,v 1.71 2000/11/08 22:09:58 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -88,6 +88,34 @@ makeRangeTable(ParseState *pstate, List *frmList)
 }
 
 /*
+ * lockTargetTable
+ *	  Find the target relation of INSERT/UPDATE/DELETE and acquire write
+ *	  lock on it.  This must be done before building the range table,
+ *	  in case the target is also mentioned as a source relation --- we
+ *	  want to be sure to grab the write lock before any read lock.
+ *
+ * The ParseState's link to the target relcache entry is also set here.
+ */
+void
+lockTargetTable(ParseState *pstate, char *relname)
+{
+	/* Close old target; this could only happen for multi-action rules */
+	if (pstate->p_target_relation != NULL)
+		heap_close(pstate->p_target_relation, NoLock);
+	pstate->p_target_relation = NULL;
+	pstate->p_target_rangetblentry = NULL; /* setTargetTable will set this */
+
+	/*
+	 * Open target rel and grab suitable lock (which we will hold till
+	 * end of transaction).
+	 *
+	 * analyze.c will eventually do the corresponding heap_close(),
+	 * but *not* release the lock.
+	 */
+	pstate->p_target_relation = heap_openr(relname, RowExclusiveLock);
+}
+
+/*
  * setTargetTable
  *	  Add the target relation of INSERT/UPDATE/DELETE to the range table,
  *	  and make the special links to it in the ParseState.
@@ -133,13 +161,10 @@ setTargetTable(ParseState *pstate, char *relname, bool inh, bool inJoinSet)
 	if (inJoinSet)
 		addRTEtoJoinList(pstate, rte);
 
-	/* This could only happen for multi-action rules */
-	if (pstate->p_target_relation != NULL)
-		heap_close(pstate->p_target_relation, AccessShareLock);
+	/* lockTargetTable should have been called earlier */
+	Assert(pstate->p_target_relation != NULL);
 
 	pstate->p_target_rangetblentry = rte;
-	pstate->p_target_relation = heap_open(rte->relid, AccessShareLock);
-	/* will close relation later, see analyze.c */
 }
 
 
diff --git a/src/backend/parser/parse_relation.c b/src/backend/parser/parse_relation.c
index 3fccd95cb18..984485f9b45 100644
--- a/src/backend/parser/parse_relation.c
+++ b/src/backend/parser/parse_relation.c
@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/parser/parse_relation.c,v 1.49 2000/09/29 18:21:36 tgl Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/parser/parse_relation.c,v 1.50 2000/11/08 22:09:58 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -34,6 +34,7 @@ static Node *scanRTEForColumn(ParseState *pstate, RangeTblEntry *rte,
 							  char *colname);
 static Node *scanJoinForColumn(JoinExpr *join, char *colname,
 							   int sublevels_up);
+static bool isForUpdate(ParseState *pstate, char *relname);
 static List *expandNamesVars(ParseState *pstate, List *names, List *vars);
 static void warnAutoRange(ParseState *pstate, char *refname);
 
@@ -477,6 +478,7 @@ addRangeTableEntry(ParseState *pstate,
 				   bool inFromCl)
 {
 	char	   *refname = alias ? alias->relname : relname;
+	LOCKMODE	lockmode;
 	Relation	rel;
 	RangeTblEntry *rte;
 	Attr	   *eref;
@@ -502,17 +504,22 @@ addRangeTableEntry(ParseState *pstate,
 
 	/*
 	 * Get the rel's OID.  This access also ensures that we have an
-	 * up-to-date relcache entry for the rel.  We don't need to keep it
-	 * open, however. Since this is open anyway, let's check that the
-	 * number of column aliases is reasonable. - Thomas 2000-02-04
+	 * up-to-date relcache entry for the rel.  Since this is typically
+	 * the first access to a rel in a statement, be careful to get the
+	 * right access level depending on whether we're doing SELECT FOR UPDATE.
 	 */
-	rel = heap_openr(relname, AccessShareLock);
+	lockmode = isForUpdate(pstate, relname) ? RowShareLock : AccessShareLock;
+	rel = heap_openr(relname, lockmode);
 	rte->relid = RelationGetRelid(rel);
-	maxattrs = RelationGetNumberOfAttributes(rel);
 
 	eref = alias ? (Attr *) copyObject(alias) : makeAttr(refname, NULL);
 	numaliases = length(eref->attrs);
 
+	/*
+	 * Since the rel is open anyway, let's check that the
+	 * number of column aliases is reasonable. - Thomas 2000-02-04
+	 */
+	maxattrs = RelationGetNumberOfAttributes(rel);
 	if (maxattrs < numaliases)
 		elog(ERROR, "Table \"%s\" has %d columns available but %d columns specified",
 			 refname, maxattrs, numaliases);
@@ -527,7 +534,12 @@ addRangeTableEntry(ParseState *pstate,
 	}
 	rte->eref = eref;
 
-	heap_close(rel, AccessShareLock);
+	/*
+	 * Drop the rel refcount, but keep the access lock till end of transaction
+	 * so that the table can't be deleted or have its schema modified
+	 * underneath us.
+	 */
+	heap_close(rel, NoLock);
 
 	/*----------
 	 * Flags:
@@ -644,6 +656,41 @@ addRangeTableEntryForSubquery(ParseState *pstate,
 }
 
 /*
+ * Has the specified relname been selected FOR UPDATE?
+ */
+static bool
+isForUpdate(ParseState *pstate, char *relname)
+{
+	/* Outer loop to check parent query levels as well as this one */
+	while (pstate != NULL)
+	{
+		if (pstate->p_forUpdate != NIL)
+		{
+			if (lfirst(pstate->p_forUpdate) == NULL)
+			{
+				/* all tables used in query */
+				return true;
+			}
+			else
+			{
+				/* just the named tables */
+				List   *l;
+
+				foreach(l, pstate->p_forUpdate)
+				{
+					char	   *rname = lfirst(l);
+
+					if (strcmp(relname, rname) == 0)
+						return true;
+				}
+			}
+		}
+		pstate = pstate->parentParseState;
+	}
+	return false;
+}
+
+/*
  * Add the given RTE as a top-level entry in the pstate's join list,
  * unless there already is an entry for it.
  */