Extend CREATE DATABASE to allow selection of a template database to be

cloned, rather than always cloning template1.  Modify initdb to generate
two identical databases rather than one, template0 and template1.
Connections to template0 are disallowed, so that it will always remain
in its virgin as-initdb'd state.  pg_dumpall now dumps databases with
restore commands that say CREATE DATABASE foo WITH TEMPLATE = template0.
This allows proper behavior when there is user-added data in template1.
initdb forced!

10 files changed, 329 insertions, 205 deletions

diff --git a/src/backend/commands/dbcommands.c b/src/backend/commands/dbcommands.c
index 2442679250f..70fd952db63 100644
--- a/src/backend/commands/dbcommands.c
+++ b/src/backend/commands/dbcommands.c
@@ -8,12 +8,11 @@
  *
  *
  * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/commands/dbcommands.c,v 1.66 2000/11/12 20:51:50 tgl Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/commands/dbcommands.c,v 1.67 2000/11/14 18:37:41 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
 #include "postgres.h"
-#include "commands/dbcommands.h"
 
 #include <errno.h>
 #include <fcntl.h>
@@ -27,6 +26,7 @@
 #include "catalog/pg_database.h"
 #include "catalog/pg_shadow.h"
 #include "commands/comment.h"
+#include "commands/dbcommands.h"
 #include "miscadmin.h"
 #include "storage/sinval.h"		/* for DatabaseHasActiveBackends */
 #include "utils/builtins.h"
@@ -35,29 +35,40 @@
 
 
 /* non-export function prototypes */
+static bool get_db_info(const char *name, Oid *dbIdP, int4 *ownerIdP,
+						int *encodingP, bool *dbIsTemplateP,
+						Oid *dbLastSysOidP, char *dbpath);
 static bool get_user_info(Oid use_sysid, bool *use_super, bool *use_createdb);
-static bool get_db_info(const char *name, char *dbpath, Oid *dbIdP, int4 *ownerIdP);
-static char * resolve_alt_dbpath(const char * dbpath, Oid dboid);
-static bool remove_dbdirs(const char * real_loc, const char * altloc);
+static char *resolve_alt_dbpath(const char *dbpath, Oid dboid);
+static bool remove_dbdirs(const char *real_loc, const char *altloc);
 
 /*
  * CREATE DATABASE
  */
 
 void
-createdb(const char *dbname, const char *dbpath, int encoding)
+createdb(const char *dbname, const char *dbpath,
+		 const char *dbtemplate, int encoding)
 {
+	char	   *nominal_loc;
+	char	   *alt_loc;
+	char	   *target_dir;
+	char		src_loc[MAXPGPATH];
 	char		buf[2 * MAXPGPATH + 100];
-	char	   *altloc;
-	char	   *real_loc;
 	int			ret;
 	bool		use_super,
 				use_createdb;
+	Oid			src_dboid;
+	int4		src_owner;
+	int			src_encoding;
+	bool		src_istemplate;
+	Oid			src_lastsysoid;
+	char		src_dbpath[MAXPGPATH];
 	Relation	pg_database_rel;
 	HeapTuple	tuple;
 	TupleDesc	pg_database_dsc;
 	Datum		new_record[Natts_pg_database];
-	char		new_record_nulls[Natts_pg_database] = {' ', ' ', ' ', ' ', ' '};
+	char		new_record_nulls[Natts_pg_database];
 	Oid			dboid;
 
 	if (!get_user_info(GetUserId(), &use_super, &use_createdb))
@@ -66,122 +77,195 @@ createdb(const char *dbname, const char *dbpath, int encoding)
 	if (!use_createdb && !use_super)
 		elog(ERROR, "CREATE DATABASE: permission denied");
 
-	if (get_db_info(dbname, NULL, NULL, NULL))
-		elog(ERROR, "CREATE DATABASE: database \"%s\" already exists", dbname);
-
 	/* don't call this in a transaction block */
 	if (IsTransactionBlock())
 		elog(ERROR, "CREATE DATABASE: may not be called in a transaction block");
 
 	/*
-	 * Insert a new tuple into pg_database
+	 * Check for db name conflict.  There is a race condition here, since
+	 * another backend could create the same DB name before we commit.
+	 * However, holding an exclusive lock on pg_database for the whole time
+	 * we are copying the source database doesn't seem like a good idea,
+	 * so accept possibility of race to create.  We will check again after
+	 * we grab the exclusive lock.
 	 */
-	pg_database_rel = heap_openr(DatabaseRelationName, AccessExclusiveLock);
-	pg_database_dsc = RelationGetDescr(pg_database_rel);
+	if (get_db_info(dbname, NULL, NULL, NULL, NULL, NULL, NULL))
+		elog(ERROR, "CREATE DATABASE: database \"%s\" already exists", dbname);
 
-	/* 
-	 * Preassign OID for pg_database tuple, so that we know current
-	 * OID counter value
+	/*
+	 * Lookup database (template) to be cloned.
 	 */
-	dboid = newoid();
+	if (!dbtemplate)
+		dbtemplate = "template1"; /* Default template database name */
 
-	/* Form tuple */
-	new_record[Anum_pg_database_datname - 1] =
-		DirectFunctionCall1(namein, CStringGetDatum(dbname));
-	new_record[Anum_pg_database_datdba - 1] = Int32GetDatum(GetUserId());
-	new_record[Anum_pg_database_encoding - 1] = Int32GetDatum(encoding);
-	/* Save current OID val */
-	new_record[Anum_pg_database_datlastsysoid - 1] = ObjectIdGetDatum(dboid);
-	/* no nulls here, GetRawDatabaseInfo doesn't like them */
-	new_record[Anum_pg_database_datpath - 1] =
-		DirectFunctionCall1(textin, CStringGetDatum(dbpath ? dbpath : ""));
+	if (!get_db_info(dbtemplate, &src_dboid, &src_owner, &src_encoding,
+					 &src_istemplate, &src_lastsysoid, src_dbpath))
+		elog(ERROR, "CREATE DATABASE: template \"%s\" does not exist",
+			 dbtemplate);
+	/*
+	 * Permission check: to copy a DB that's not marked datistemplate,
+	 * you must be superuser or the owner thereof.
+	 */
+	if (!src_istemplate)
+	{
+		if (!use_super && GetUserId() != src_owner)
+			elog(ERROR, "CREATE DATABASE: permission to copy \"%s\" denied",
+				 dbtemplate);
+	}
+	/*
+	 * Determine physical path of source database
+	 */
+	alt_loc = resolve_alt_dbpath(src_dbpath, src_dboid);
+	if (!alt_loc)
+		alt_loc = GetDatabasePath(src_dboid);
+	strcpy(src_loc, alt_loc);
 
-	tuple = heap_formtuple(pg_database_dsc, new_record, new_record_nulls);
+	/*
+	 * The source DB can't have any active backends, except this one
+	 * (exception is to allow CREATE DB while connected to template1).
+	 * Otherwise we might copy inconsistent data.  This check is not
+	 * bulletproof, since someone might connect while we are copying...
+	 */
+	if (DatabaseHasActiveBackends(src_dboid, true))
+		elog(ERROR, "CREATE DATABASE: source database \"%s\" is being accessed by other users", dbtemplate);
 
-	tuple->t_data->t_oid = dboid;	/* override heap_insert */
+	/* If encoding is defaulted, use source's encoding */
+	if (encoding < 0)
+		encoding = src_encoding;
 
+	/* 
+	 * Preassign OID for pg_database tuple, so that we can compute db path.
+	 */
+	dboid = newoid();
 
 	/*
-	 * Update table
+	 * Compute nominal location (where we will try to access the database),
+	 * and resolve alternate physical location if one is specified.
 	 */
-	heap_insert(pg_database_rel, tuple);
-
-	real_loc = GetDatabasePath(tuple->t_data->t_oid);
-	altloc = resolve_alt_dbpath(dbpath, tuple->t_data->t_oid);
+	nominal_loc = GetDatabasePath(dboid);
+	alt_loc = resolve_alt_dbpath(dbpath, dboid);
 
-	if (strchr(real_loc, '\'') && strchr(altloc, '\''))
+	if (strchr(nominal_loc, '\''))
+		elog(ERROR, "database path may not contain single quotes");
+	if (alt_loc && strchr(alt_loc, '\''))
+		elog(ERROR, "database path may not contain single quotes");
+	if (strchr(src_loc, '\''))
 		elog(ERROR, "database path may not contain single quotes");
 	/* ... otherwise we'd be open to shell exploits below */
 
-	/*
-	 * Update indexes (there aren't any currently)
-	 */
-#ifdef Num_pg_database_indices
-	if (RelationGetForm(pg_database_rel)->relhasindex)
-	{
-		Relation	idescs[Num_pg_database_indices];
-
-		CatalogOpenIndices(Num_pg_database_indices,
-						   Name_pg_database_indices, idescs);
-		CatalogIndexInsert(idescs, Num_pg_database_indices, pg_database_rel,
-						   tuple);
-		CatalogCloseIndices(Num_pg_database_indices, idescs);
-	}
+#ifdef XLOG
+	/* Try to force any dirty buffers out to disk */
+	BufferSync();
 #endif
 
-	heap_close(pg_database_rel, NoLock);
-
 	/*
 	 * Close virtual file descriptors so the kernel has more available for
 	 * the mkdir() and system() calls below.
 	 */
 	closeAllVfds();
 
-	/* Copy the template database to the new location */
+	/*
+	 * Check we can create the target directory --- but then remove it
+	 * because we rely on cp(1) to create it for real.
+	 */
+	target_dir = alt_loc ? alt_loc : nominal_loc;
 
-	if (mkdir((altloc ? altloc : real_loc), S_IRWXU) != 0)
-		elog(ERROR, "CREATE DATABASE: unable to create database directory '%s': %s",
-			 (altloc ? altloc : real_loc), strerror(errno));
+	if (mkdir(target_dir, S_IRWXU) != 0)
+		elog(ERROR, "CREATE DATABASE: unable to create database directory '%s': %m",
+			 target_dir);
+	rmdir(target_dir);
 
-	if (altloc)
+	/* Make the symlink, if needed */
+	if (alt_loc)
 	{
-		if (symlink(altloc, real_loc) != 0)
-			elog(ERROR, "CREATE DATABASE: could not link %s to %s: %s",
-				 real_loc, altloc, strerror(errno));
+		if (symlink(alt_loc, nominal_loc) != 0)
+			elog(ERROR, "CREATE DATABASE: could not link '%s' to '%s': %m",
+				 nominal_loc, alt_loc);
 	}
 
-	snprintf(buf, sizeof(buf), "cp '%s'/* '%s'",
-			 GetDatabasePath(TemplateDbOid), real_loc);
+	/* Copy the template database to the new location */
+	snprintf(buf, sizeof(buf), "cp -r '%s' '%s'", src_loc, target_dir);
 
 	ret = system(buf);
 	/* Some versions of SunOS seem to return ECHILD after a system() call */
 	if (ret != 0 && errno != ECHILD)
 	{
-		if (remove_dbdirs(real_loc, altloc))
+		if (remove_dbdirs(nominal_loc, alt_loc))
 			elog(ERROR, "CREATE DATABASE: could not initialize database directory");
 		else
 			elog(ERROR, "CREATE DATABASE: could not initialize database directory; delete failed as well");
 	}
 
-#ifdef XLOG
-	BufferSync();
+	/*
+	 * Now OK to grab exclusive lock on pg_database.
+	 */
+	pg_database_rel = heap_openr(DatabaseRelationName, AccessExclusiveLock);
+
+	/* Check to see if someone else created same DB name meanwhile. */
+	if (get_db_info(dbname, NULL, NULL, NULL, NULL, NULL, NULL))
+	{
+		remove_dbdirs(nominal_loc, alt_loc);
+		elog(ERROR, "CREATE DATABASE: database \"%s\" already exists", dbname);
+	}
+
+	/*
+	 * Insert a new tuple into pg_database
+	 */
+	pg_database_dsc = RelationGetDescr(pg_database_rel);
+
+	/* Form tuple */
+	new_record[Anum_pg_database_datname - 1] =
+		DirectFunctionCall1(namein, CStringGetDatum(dbname));
+	new_record[Anum_pg_database_datdba - 1] = Int32GetDatum(GetUserId());
+	new_record[Anum_pg_database_encoding - 1] = Int32GetDatum(encoding);
+	new_record[Anum_pg_database_datistemplate - 1] = BoolGetDatum(false);
+	new_record[Anum_pg_database_datallowconn - 1] = BoolGetDatum(true);
+	new_record[Anum_pg_database_datlastsysoid - 1] = ObjectIdGetDatum(src_lastsysoid);
+	/* no nulls here, GetRawDatabaseInfo doesn't like them */
+	new_record[Anum_pg_database_datpath - 1] =
+		DirectFunctionCall1(textin, CStringGetDatum(dbpath ? dbpath : ""));
+
+	memset(new_record_nulls, ' ', sizeof(new_record_nulls));
+
+	tuple = heap_formtuple(pg_database_dsc, new_record, new_record_nulls);
+
+	tuple->t_data->t_oid = dboid;	/* override heap_insert's OID selection */
+
+	heap_insert(pg_database_rel, tuple);
+
+	/*
+	 * Update indexes (there aren't any currently)
+	 */
+#ifdef Num_pg_database_indices
+	if (RelationGetForm(pg_database_rel)->relhasindex)
+	{
+		Relation	idescs[Num_pg_database_indices];
+
+		CatalogOpenIndices(Num_pg_database_indices,
+						   Name_pg_database_indices, idescs);
+		CatalogIndexInsert(idescs, Num_pg_database_indices, pg_database_rel,
+						   tuple);
+		CatalogCloseIndices(Num_pg_database_indices, idescs);
+	}
 #endif
-}
 
+	/* Close pg_database, but keep lock till commit */
+	heap_close(pg_database_rel, NoLock);
+}
 
 
 /*
  * DROP DATABASE
  */
-
 void
 dropdb(const char *dbname)
 {
 	int4		db_owner;
+	bool		db_istemplate;
 	bool		use_super;
 	Oid			db_id;
-	char       *altloc;
-	char       *real_loc;
+	char       *alt_loc;
+	char       *nominal_loc;
 	char        dbpath[MAXPGPATH];
 	Relation	pgdbrel;
 	HeapScanDesc pgdbscan;
@@ -190,9 +274,6 @@ dropdb(const char *dbname)
 
 	AssertArg(dbname);
 
-	if (strcmp(dbname, "template1") == 0)
-		elog(ERROR, "DROP DATABASE: may not be executed on the template1 database");
-
 	if (strcmp(dbname, DatabaseName) == 0)
 		elog(ERROR, "DROP DATABASE: cannot be executed on the currently open database");
 
@@ -202,15 +283,6 @@ dropdb(const char *dbname)
 	if (!get_user_info(GetUserId(), &use_super, NULL))
 		elog(ERROR, "current user name is invalid");
 
-	if (!get_db_info(dbname, dbpath, &db_id, &db_owner))
-		elog(ERROR, "DROP DATABASE: database \"%s\" does not exist", dbname);
-
-	if (GetUserId() != db_owner && !use_super)
-		elog(ERROR, "DROP DATABASE: permission denied");
-
-	real_loc = GetDatabasePath(db_id);
-	altloc = resolve_alt_dbpath(dbpath, db_id);
-
 	/*
 	 * Obtain exclusive lock on pg_database.  We need this to ensure that
 	 * no new backend starts up in the target database while we are
@@ -222,14 +294,29 @@ dropdb(const char *dbname)
 	 */
 	pgdbrel = heap_openr(DatabaseRelationName, AccessExclusiveLock);
 
+	if (!get_db_info(dbname, &db_id, &db_owner, NULL,
+					 &db_istemplate, NULL, dbpath))
+		elog(ERROR, "DROP DATABASE: database \"%s\" does not exist", dbname);
+
+	if (!use_super && GetUserId() != db_owner)
+		elog(ERROR, "DROP DATABASE: permission denied");
+
+	/*
+	 * Disallow dropping a DB that is marked istemplate.  This is just
+	 * to prevent people from accidentally dropping template0 or template1;
+	 * they can do so if they're really determined ...
+	 */
+	if (db_istemplate)
+		elog(ERROR, "DROP DATABASE: database is marked as a template");
+
+	nominal_loc = GetDatabasePath(db_id);
+	alt_loc = resolve_alt_dbpath(dbpath, db_id);
+
 	/*
 	 * Check for active backends in the target database.
 	 */
 	if (DatabaseHasActiveBackends(db_id, false))
-	{
-		heap_close(pgdbrel, AccessExclusiveLock);
 		elog(ERROR, "DROP DATABASE: database \"%s\" is being accessed by other users", dbname);
-	}
 
 	/*
 	 * Find the database's tuple by OID (should be unique, we trust).
@@ -242,8 +329,6 @@ dropdb(const char *dbname)
 	tup = heap_getnext(pgdbscan, 0);
 	if (!HeapTupleIsValid(tup))
 	{
-		heap_close(pgdbrel, AccessExclusiveLock);
-
 		/*
 		 * This error should never come up since the existence of the
 		 * database is checked earlier
@@ -252,9 +337,6 @@ dropdb(const char *dbname)
 			 dbname);
 	}
 
-	/* Delete any comments associated with the database */
-	DeleteComments(db_id);
-
 	/* Remove the database's tuple from pg_database */
 	heap_delete(pgdbrel, &tup->t_self, NULL);
 
@@ -266,6 +348,9 @@ dropdb(const char *dbname)
 	 */
 	heap_close(pgdbrel, NoLock);
 
+	/* Delete any comments associated with the database */
+	DeleteComments(db_id);
+
 	/*
 	 * Drop pages for this database that are in the shared buffer cache.
 	 * This is important to ensure that no remaining backend tries to
@@ -274,15 +359,9 @@ dropdb(const char *dbname)
 	DropBuffers(db_id);
 
 	/*
-	 * Close virtual file descriptors so the kernel has more available for
-	 * the system() call below.
-	 */
-	closeAllVfds();
-
-	/*
 	 * Remove the database's subdirectory and everything in it.
 	 */
-	remove_dbdirs(real_loc, altloc);
+	remove_dbdirs(nominal_loc, alt_loc);
 }
 
 
@@ -292,28 +371,32 @@ dropdb(const char *dbname)
  */
 
 static bool
-get_db_info(const char *name, char *dbpath, Oid *dbIdP, int4 *ownerIdP)
+get_db_info(const char *name, Oid *dbIdP, int4 *ownerIdP,
+			int *encodingP, bool *dbIsTemplateP,
+			Oid *dbLastSysOidP, char *dbpath)
 {
 	Relation	relation;
-	HeapTuple	tuple;
 	ScanKeyData scanKey;
 	HeapScanDesc scan;
+	HeapTuple	tuple;
 
 	AssertArg(name);
 
-	relation = heap_openr(DatabaseRelationName, AccessExclusiveLock /* ??? */ );
+	/* Caller may wish to grab a better lock on pg_database beforehand... */
+	relation = heap_openr(DatabaseRelationName, AccessShareLock);
 
 	ScanKeyEntryInitialize(&scanKey, 0, Anum_pg_database_datname,
 						   F_NAMEEQ, NameGetDatum(name));
 
 	scan = heap_beginscan(relation, 0, SnapshotNow, 1, &scanKey);
 	if (!HeapScanIsValid(scan))
-		elog(ERROR, "Cannot begin scan of %s.", DatabaseRelationName);
+		elog(ERROR, "Cannot begin scan of %s", DatabaseRelationName);
 
 	tuple = heap_getnext(scan, 0);
 
 	if (HeapTupleIsValid(tuple))
 	{
+		Form_pg_database dbform = (Form_pg_database) GETSTRUCT(tuple);
 		text	   *tmptext;
 		bool		isnull;
 
@@ -322,22 +405,23 @@ get_db_info(const char *name, char *dbpath, Oid *dbIdP, int4 *ownerIdP)
 			*dbIdP = tuple->t_data->t_oid;
 		/* uid of the owner */
 		if (ownerIdP)
-		{
-			*ownerIdP = (int4) heap_getattr(tuple,
-											Anum_pg_database_datdba,
-											RelationGetDescr(relation),
-											&isnull);
-			if (isnull)
-				*ownerIdP = -1; /* hopefully no one has that id already ;) */
-		}
+			*ownerIdP = dbform->datdba;
+		/* multibyte encoding */
+		if (encodingP)
+			*encodingP = dbform->encoding;
+		/* allowed as template? */
+		if (dbIsTemplateP)
+			*dbIsTemplateP = dbform->datistemplate;
+		/* last system OID used in database */
+		if (dbLastSysOidP)
+			*dbLastSysOidP = dbform->datlastsysoid;
 		/* database path (as registered in pg_database) */
 		if (dbpath)
 		{
-			tmptext = (text *) heap_getattr(tuple,
-											Anum_pg_database_datpath,
-											RelationGetDescr(relation),
-											&isnull);
-
+			tmptext = DatumGetTextP(heap_getattr(tuple,
+												 Anum_pg_database_datpath,
+												 RelationGetDescr(relation),
+												 &isnull));
 			if (!isnull)
 			{
 				Assert(VARSIZE(tmptext) - VARHDRSZ < MAXPGPATH);
@@ -349,16 +433,9 @@ get_db_info(const char *name, char *dbpath, Oid *dbIdP, int4 *ownerIdP)
 				strcpy(dbpath, "");
 		}
 	}
-	else
-	{
-		if (dbIdP)
-			*dbIdP = InvalidOid;
-	}
 
 	heap_endscan(scan);
-
-	/* We will keep the lock on the relation until end of transaction. */
-	heap_close(relation, NoLock);
+	heap_close(relation, AccessShareLock);
 
 	return HeapTupleIsValid(tuple);
 }
@@ -396,6 +473,8 @@ resolve_alt_dbpath(const char * dbpath, Oid dboid)
 
 	if (strchr(dbpath, '/'))
 	{
+		if (dbpath[0] != '/')
+			elog(ERROR, "Relative paths are not allowed as database locations");
 #ifndef ALLOW_ABSOLUTE_DBPATHS
 		elog(ERROR, "Absolute paths are not allowed as database locations");
 #endif
@@ -406,9 +485,9 @@ resolve_alt_dbpath(const char * dbpath, Oid dboid)
 		/* must be environment variable */
 		char * var = getenv(dbpath);
 		if (!var)
-			elog(ERROR, "environment variable %s not set", dbpath);
+			elog(ERROR, "Postmaster environment variable '%s' not set", dbpath);
 		if (var[0] != '/')
-			elog(ERROR, "environment variable %s must be absolute path", dbpath);
+			elog(ERROR, "Postmaster environment variable '%s' must be absolute path", dbpath);
 		prefix = var;
 	}
 
@@ -421,24 +500,36 @@ resolve_alt_dbpath(const char * dbpath, Oid dboid)
 
 
 static bool
-remove_dbdirs(const char * real_loc, const char * altloc)
+remove_dbdirs(const char * nominal_loc, const char * alt_loc)
 {
+	const char   *target_dir;
 	char buf[MAXPGPATH + 100];
 	bool success = true;
 
-	if (altloc)
+	target_dir = alt_loc ? alt_loc : nominal_loc;
+
+	/*
+	 * Close virtual file descriptors so the kernel has more available for
+	 * the system() call below.
+	 */
+	closeAllVfds();
+
+	if (alt_loc)
+	{
 		/* remove symlink */
-		if (unlink(real_loc) != 0)
+		if (unlink(nominal_loc) != 0)
 		{
-			elog(NOTICE, "could not remove '%s': %s", real_loc, strerror(errno));
+			elog(NOTICE, "could not remove '%s': %m", nominal_loc);
 			success = false;
 		}
+	}
+
+	snprintf(buf, sizeof(buf), "rm -rf '%s'", target_dir);
 
-	snprintf(buf, sizeof(buf), "rm -rf '%s'", altloc ? altloc : real_loc);
 	if (system(buf) != 0 && errno != ECHILD)
 	{
 		elog(NOTICE, "database directory '%s' could not be removed",
-			 altloc ? altloc : real_loc);
+			 target_dir);
 		success = false;
 	}
 
diff --git a/src/backend/nodes/copyfuncs.c b/src/backend/nodes/copyfuncs.c
index 3490ef80624..9c342b5dc1e 100644
--- a/src/backend/nodes/copyfuncs.c
+++ b/src/backend/nodes/copyfuncs.c
@@ -15,7 +15,7 @@
  * Portions Copyright (c) 1994, Regents of the University of California
  *
  * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/nodes/copyfuncs.c,v 1.131 2000/11/12 00:36:57 tgl Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/nodes/copyfuncs.c,v 1.132 2000/11/14 18:37:42 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -2231,6 +2231,8 @@ _copyCreatedbStmt(CreatedbStmt *from)
 		newnode->dbname = pstrdup(from->dbname);
 	if (from->dbpath)
 		newnode->dbpath = pstrdup(from->dbpath);
+	if (from->dbtemplate)
+		newnode->dbtemplate = pstrdup(from->dbtemplate);
 	newnode->encoding = from->encoding;
 
 	return newnode;
diff --git a/src/backend/nodes/equalfuncs.c b/src/backend/nodes/equalfuncs.c
index 8fb81a45c0b..aac0902a028 100644
--- a/src/backend/nodes/equalfuncs.c
+++ b/src/backend/nodes/equalfuncs.c
@@ -20,7 +20,7 @@
  * Portions Copyright (c) 1994, Regents of the University of California
  *
  * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/nodes/equalfuncs.c,v 1.81 2000/11/12 00:36:57 tgl Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/nodes/equalfuncs.c,v 1.82 2000/11/14 18:37:42 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -1087,6 +1087,8 @@ _equalCreatedbStmt(CreatedbStmt *a, CreatedbStmt *b)
 		return false;
 	if (!equalstr(a->dbpath, b->dbpath))
 		return false;
+	if (!equalstr(a->dbtemplate, b->dbtemplate))
+		return false;
 	if (a->encoding != b->encoding)
 		return false;
 
diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y
index 5572828d259..3c4a2e00c9c 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.208 2000/11/08 22:09:58 tgl Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/parser/gram.y,v 2.209 2000/11/14 18:37:49 tgl Exp $
  *
  * HISTORY
  *	  AUTHOR			DATE			MAJOR EVENT
@@ -56,8 +56,8 @@
 #include "miscadmin.h"
 #include "mb/pg_wchar.h"
 #else
-#define GetTemplateEncoding()	0		/* SQL_ASCII */
-#define GetTemplateEncodingName()	"SQL_ASCII"
+#define GetStandardEncoding()	0		/* SQL_ASCII */
+#define GetStandardEncodingName()	"SQL_ASCII"
 #endif
 
 extern List *parsetree;			/* final parse result is delivered here */
@@ -146,8 +146,7 @@ static void doNegateFloat(Value *v);
 %type <node>    alter_column_action
 %type <ival>    drop_behavior
 
-%type <str>		createdb_opt_location
-%type <ival>    createdb_opt_encoding
+%type <list>	createdb_opt_list, createdb_opt_item
 
 %type <ival>	opt_lock, lock_type
 %type <boolean>	opt_lmode, opt_force
@@ -347,7 +346,7 @@ static void doNegateFloat(Value *v);
 		OFFSET, OIDS, OPERATOR, OWNER, PASSWORD, PROCEDURAL,
 		REINDEX, RENAME, RESET, RETURNS, ROW, RULE,
 		SEQUENCE, SERIAL, SETOF, SHARE, SHOW, START, STATEMENT, STDIN, STDOUT, SYSID,
-		TEMP, TOAST, TRUNCATE, TRUSTED, 
+		TEMP, TEMPLATE, TOAST, TRUNCATE, TRUSTED, 
 		UNLISTEN, UNTIL, VACUUM, VALID, VERBOSE, VERSION
 
 /* The grammar thinks these are keywords, but they are not in the keywords.c
@@ -687,7 +686,8 @@ CreateSchemaStmt:  CREATE SCHEMA UserId
 					CreatedbStmt *n = makeNode(CreatedbStmt);
 					n->dbname = $3;
 					n->dbpath = NULL;
-					n->encoding = GetTemplateEncoding();
+					n->dbtemplate = NULL;
+					n->encoding = -1;
 					$$ = (Node *)n;
 				}
 		;
@@ -2924,16 +2924,34 @@ LoadStmt:  LOAD file_name
  *
  *****************************************************************************/
 
-CreatedbStmt:  CREATE DATABASE database_name WITH createdb_opt_location createdb_opt_encoding
+CreatedbStmt:  CREATE DATABASE database_name WITH createdb_opt_list
 				{
 					CreatedbStmt *n = makeNode(CreatedbStmt);
-
-					if ($5 == NULL && $6 == -1)
-						elog(ERROR, "CREATE DATABASE WITH requires at least one option");
+					List   *l;
 
 					n->dbname = $3;
-					n->dbpath = $5;
-					n->encoding = ($6 == -1) ? GetTemplateEncoding() : $6;
+					/* set default options */
+					n->dbpath = NULL;
+					n->dbtemplate = NULL;
+					n->encoding = -1;
+					/* process additional options */
+					foreach(l, $5)
+					{
+						List   *optitem = (List *) lfirst(l);
+
+						switch (lfirsti(optitem))
+						{
+							case 1:
+								n->dbpath = (char *) lsecond(optitem);
+								break;
+							case 2:
+								n->dbtemplate = (char *) lsecond(optitem);
+								break;
+							case 3:
+								n->encoding = lfirsti(lnext(optitem));
+								break;
+						}
+					}
 					$$ = (Node *)n;
 				}
 		| CREATE DATABASE database_name
@@ -2941,27 +2959,51 @@ CreatedbStmt:  CREATE DATABASE database_name WITH createdb_opt_location createdb
 					CreatedbStmt *n = makeNode(CreatedbStmt);
 					n->dbname = $3;
 					n->dbpath = NULL;
-					n->encoding = GetTemplateEncoding();
+					n->dbtemplate = NULL;
+					n->encoding = -1;
 					$$ = (Node *)n;
 				}
 		;
 
-createdb_opt_location:  LOCATION '=' Sconst		{ $$ = $3; }
-		| LOCATION '=' DEFAULT					{ $$ = NULL; }
-		| /*EMPTY*/								{ $$ = NULL; }
+createdb_opt_list:  createdb_opt_item
+				{ $$ = makeList1($1); }
+		| createdb_opt_list createdb_opt_item
+				{ $$ = lappend($1, $2); }
 		;
 
-createdb_opt_encoding:  ENCODING '=' Sconst
+/*
+ * createdb_opt_item returns 2-element lists, with the first element
+ * being an integer code to indicate which item was specified.
+ */
+createdb_opt_item:  LOCATION '=' Sconst
 				{
+					$$ = lconsi(1, makeList1($3));
+				}
+		| LOCATION '=' DEFAULT
+				{
+					$$ = lconsi(1, makeList1((char *) NULL));
+				}
+		| TEMPLATE '=' name
+				{
+					$$ = lconsi(2, makeList1($3));
+				}
+		| TEMPLATE '=' DEFAULT
+				{
+					$$ = lconsi(2, makeList1((char *) NULL));
+				}
+		| ENCODING '=' Sconst
+				{
+					int		encoding;
 #ifdef MULTIBYTE
-					$$ = pg_char_to_encoding($3);
-					if ($$ == -1)
+					encoding = pg_char_to_encoding($3);
+					if (encoding == -1)
 						elog(ERROR, "%s is not a valid encoding name", $3);
 #else
-					if (strcasecmp($3, GetTemplateEncodingName()) != 0)
+					if (strcasecmp($3, GetStandardEncodingName()) != 0)
 						elog(ERROR, "Multi-byte support is not enabled");
-					$$ = GetTemplateEncoding();
+					encoding = GetStandardEncoding();
 #endif
+					$$ = lconsi(3, makeListi1(encoding));
 				}
 		| ENCODING '=' Iconst
 				{
@@ -2969,18 +3011,14 @@ createdb_opt_encoding:  ENCODING '=' Sconst
 					if (!pg_get_encent_by_encoding($3))
 						elog(ERROR, "%d is not a valid encoding code", $3);
 #else
-					if ($3 != GetTemplateEncoding())
+					if ($3 != GetStandardEncoding())
 						elog(ERROR, "Multi-byte support is not enabled");
 #endif
-					$$ = $3;
+					$$ = lconsi(3, makeListi1($3));
 				}
 		| ENCODING '=' DEFAULT
 				{
-					$$ = GetTemplateEncoding();
-				}
-		| /*EMPTY*/
-				{
-					$$ = -1;
+					$$ = lconsi(3, makeListi1(-1));
 				}
 		;
 
@@ -5495,6 +5533,7 @@ TokenId:  ABSOLUTE						{ $$ = "absolute"; }
 		| STDOUT						{ $$ = "stdout"; }
 		| SYSID							{ $$ = "sysid"; }
 		| TEMP							{ $$ = "temp"; }
+		| TEMPLATE						{ $$ = "template"; }
 		| TEMPORARY						{ $$ = "temporary"; }
 		| TIMEZONE_HOUR					{ $$ = "timezone_hour"; }
 		| TIMEZONE_MINUTE				{ $$ = "timezone_minute"; }
diff --git a/src/backend/parser/keywords.c b/src/backend/parser/keywords.c
index f3e4d85e4c5..19ec40f15ba 100644
--- a/src/backend/parser/keywords.c
+++ b/src/backend/parser/keywords.c
@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/parser/keywords.c,v 1.84 2000/11/08 21:28:06 momjian Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/parser/keywords.c,v 1.85 2000/11/14 18:37:49 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -247,6 +247,7 @@ static ScanKeyword ScanKeywords[] = {
 	{"sysid", SYSID},
 	{"table", TABLE},
 	{"temp", TEMP},
+	{"template", TEMPLATE},
 	{"temporary", TEMPORARY},
 	{"then", THEN},
 	{"time", TIME},
diff --git a/src/backend/postmaster/postmaster.c b/src/backend/postmaster/postmaster.c
index 3a140488e4e..68f7c06cd0f 100644
--- a/src/backend/postmaster/postmaster.c
+++ b/src/backend/postmaster/postmaster.c
@@ -11,7 +11,7 @@
  *
  *
  * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/postmaster/postmaster.c,v 1.186 2000/11/14 18:11:31 petere Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/postmaster/postmaster.c,v 1.187 2000/11/14 18:37:42 tgl Exp $
  *
  * NOTES
  *
@@ -279,14 +279,8 @@ checkDataDir(const char *checkdir)
 		exit(2);
 	}
 
-#ifdef OLD_FILE_NAMING
-	snprintf(path, sizeof(path), "%s%cbase%ctemplate1%cpg_class",
-			 checkdir, SEP_CHAR, SEP_CHAR, SEP_CHAR);
-#else
-	snprintf(path, sizeof(path), "%s%cbase%c%u%c%u",
-			 checkdir, SEP_CHAR, SEP_CHAR, 
-			 TemplateDbOid, SEP_CHAR, RelOid_pg_class);
-#endif
+	snprintf(path, sizeof(path), "%s%cglobal%cpg_control",
+			 checkdir, SEP_CHAR, SEP_CHAR);
 
 	fp = AllocateFile(path, PG_BINARY_R);
 	if (fp == NULL)
diff --git a/src/backend/tcop/utility.c b/src/backend/tcop/utility.c
index 79ac7c0f157..63a43315286 100644
--- a/src/backend/tcop/utility.c
+++ b/src/backend/tcop/utility.c
@@ -10,7 +10,7 @@
  *
  *
  * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/tcop/utility.c,v 1.101 2000/11/08 16:31:06 tgl Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/tcop/utility.c,v 1.102 2000/11/14 18:37:43 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -604,7 +604,8 @@ ProcessUtility(Node *parsetree,
 
 				set_ps_display(commandTag = "CREATE DATABASE");
 
-				createdb(stmt->dbname, stmt->dbpath, stmt->encoding);
+				createdb(stmt->dbname, stmt->dbpath,
+						 stmt->dbtemplate, stmt->encoding);
 			}
 			break;
 
diff --git a/src/backend/utils/init/postinit.c b/src/backend/utils/init/postinit.c
index f786eb1d10d..3a9e5a1797b 100644
--- a/src/backend/utils/init/postinit.c
+++ b/src/backend/utils/init/postinit.c
@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/utils/init/postinit.c,v 1.70 2000/11/12 20:51:52 tgl Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/utils/init/postinit.c,v 1.71 2000/11/14 18:37:44 tgl Exp $
  *
  *
  *-------------------------------------------------------------------------
@@ -79,6 +79,7 @@ ReverifyMyDatabase(const char *name)
 	HeapScanDesc pgdbscan;
 	ScanKeyData key;
 	HeapTuple	tup;
+	Form_pg_database dbform;
 
 	/*
 	 * Because we grab AccessShareLock here, we can be sure that destroydb
@@ -106,25 +107,24 @@ ReverifyMyDatabase(const char *name)
 		 */
 		DropBuffers(MyDatabaseId);
 		/* Now I can commit hara-kiri with a clear conscience... */
-		elog(FATAL, "Database '%s', OID %u, has disappeared from pg_database",
+		elog(FATAL, "Database \"%s\", OID %u, has disappeared from pg_database",
 			 name, MyDatabaseId);
 	}
 
 	/*
+	 * Also check that the database is currently allowing connections.
+	 */
+	dbform = (Form_pg_database) GETSTRUCT(tup);
+	if (! dbform->datallowconn)
+		elog(FATAL, "Database \"%s\" is not currently accepting connections",
+			 name);
+
+	/*
 	 * OK, we're golden.  Only other to-do item is to save the MULTIBYTE
-	 * encoding info out of the pg_database tuple.	Note we also set the
-	 * "template encoding", which is the default encoding for any CREATE
-	 * DATABASE commands executed in this backend; essentially, you get
-	 * the same encoding of the database you connected to as the default.
-	 * (This replaces code that unreliably grabbed template1's encoding
-	 * out of pg_database.	We could do an extra scan to find template1's
-	 * tuple, but for 99.99% of all backend startups it'd be wasted cycles
-	 * --- and the 'createdb' script connects to template1 anyway, so
-	 * there's no difference.)
+	 * encoding info out of the pg_database tuple.
 	 */
 #ifdef MULTIBYTE
-	SetDatabaseEncoding(((Form_pg_database) GETSTRUCT(tup))->encoding);
-	SetTemplateEncoding(((Form_pg_database) GETSTRUCT(tup))->encoding);
+	SetDatabaseEncoding(dbform->encoding);
 #endif
 
 	heap_endscan(pgdbscan);
diff --git a/src/backend/utils/mb/mbutils.c b/src/backend/utils/mb/mbutils.c
index 7cf082dc0ce..ad1322fe52b 100644
--- a/src/backend/utils/mb/mbutils.c
+++ b/src/backend/utils/mb/mbutils.c
@@ -3,7 +3,7 @@
  * client encoding and server internal encoding.
  * (currently mule internal code (mic) is used)
  * Tatsuo Ishii
- * $Id: mbutils.c,v 1.13 2000/10/30 10:40:28 ishii Exp $ */
+ * $Id: mbutils.c,v 1.14 2000/11/14 18:37:44 tgl Exp $ */
 
 
 #include "postgres.h"
@@ -271,6 +271,7 @@ pg_mbcliplen(const unsigned char *mbstr, int len, int limit)
  * fuctions for utils/init
  */
 static int	DatabaseEncoding = MULTIBYTE;
+
 void
 SetDatabaseEncoding(int encoding)
 {
@@ -289,17 +290,3 @@ getdatabaseencoding(PG_FUNCTION_ARGS)
 {
 	PG_RETURN_NAME(pg_encoding_to_char(DatabaseEncoding));
 }
-
-/* set and get template1 database encoding */
-static int	templateEncoding;
-void
-SetTemplateEncoding(int encoding)
-{
-	templateEncoding = encoding;
-}
-
-int
-GetTemplateEncoding()
-{
-	return (templateEncoding);
-}
diff --git a/src/backend/utils/misc/database.c b/src/backend/utils/misc/database.c
index f415e5aee18..364075c8bed 100644
--- a/src/backend/utils/misc/database.c
+++ b/src/backend/utils/misc/database.c
@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/utils/misc/Attic/database.c,v 1.40 2000/10/16 14:52:19 vadim Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/utils/misc/Attic/database.c,v 1.41 2000/11/14 18:37:45 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -186,7 +186,7 @@ GetRawDatabaseInfo(const char *name, Oid *db_id, char *path)
 		max = PageGetMaxOffsetNumber(pg);
 
 		/* look at each tuple on the page */
-		for (i = 0; i <= max; i++)
+		for (i = 0; i < max; i++)
 		{
 			int			offset;
 
@@ -221,8 +221,11 @@ GetRawDatabaseInfo(const char *name, Oid *db_id, char *path)
 			 * database OID from a flat file, handled the same way we
 			 * handle the password relation?
 			 */
-			if (TransactionIdIsValid((TransactionId) tup.t_data->t_xmax))
-				continue;
+			if (tup.t_data->t_infomask & HEAP_XMIN_INVALID)
+				continue;		/* inserting xact known aborted */
+			if (TransactionIdIsValid((TransactionId) tup.t_data->t_xmax) &&
+				!(tup.t_data->t_infomask & HEAP_XMAX_INVALID))
+				continue;		/* deleting xact happened, not known aborted */
 
 			/*
 			 * Okay, see if this is the one we want.
@@ -241,6 +244,10 @@ GetRawDatabaseInfo(const char *name, Oid *db_id, char *path)
 		}
 	}
 
+	/* failed to find it... */
+	*db_id = InvalidOid;
+	*path = '\0';
+
 done:
 	close(dbfd);
 	pfree(pg);