Tablespaces. Alternate database locations are dead, long live tablespaces.

There are various things left to do: contrib dbsize and oid2name modules need work, and so does the documentation. Also someone should think about COMMENT ON TABLESPACE and maybe RENAME TABLESPACE. Also initlocation is dead, it just doesn't know it yet. Gavin Sherry and Tom Lane.
2025-11-09 06:21:09 +03:00 · 2004-06-18 06:14:31 +00:00
parent 474875f443
commit 2467394ee1
94 changed files with 3765 additions and 1588 deletions
--- a/src/backend/commands/Makefile
+++ b/src/backend/commands/Makefile
@@ -4,7 +4,7 @@
 #    Makefile for backend/commands
 #
 # IDENTIFICATION
-#    $PostgreSQL: pgsql/src/backend/commands/Makefile,v 1.33 2003/11/29 19:51:47 pgsql Exp $
+#    $PostgreSQL: pgsql/src/backend/commands/Makefile,v 1.34 2004/06/18 06:13:22 tgl Exp $
 #
 #-------------------------------------------------------------------------

@@ -17,8 +17,8 @@ OBJS = aggregatecmds.o alter.o analyze.o async.o cluster.o comment.o  \
 	dbcommands.o define.o explain.o functioncmds.o \
 	indexcmds.o lockcmds.o operatorcmds.o opclasscmds.o \
 	portalcmds.o prepare.o proclang.o \
-	schemacmds.o sequence.o tablecmds.o trigger.o typecmds.o user.o \
-	vacuum.o vacuumlazy.o variable.o view.o
+	schemacmds.o sequence.o tablecmds.o tablespace.o trigger.o \
+	typecmds.o user.o vacuum.o vacuumlazy.o variable.o view.o

 all: SUBSYS.o

--- a/src/backend/commands/cluster.c
+++ b/src/backend/commands/cluster.c
@@ -11,7 +11,7 @@
 *
 *
 * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/commands/cluster.c,v 1.125 2004/05/31 19:24:05 tgl Exp $
+ *	  $PostgreSQL: pgsql/src/backend/commands/cluster.c,v 1.126 2004/06/18 06:13:22 tgl Exp $
 *
 *-------------------------------------------------------------------------
 */
@@ -568,6 +568,7 @@ make_new_heap(Oid OIDOldHeap, const char *NewName)

 	OIDNewHeap = heap_create_with_catalog(NewName,
 										  RelationGetNamespace(OldHeap),
+		                                  OldHeap->rd_rel->reltablespace,
 										  tupdesc,
 										  OldHeap->rd_rel->relkind,
 										  OldHeap->rd_rel->relisshared,
--- a/src/backend/commands/dbcommands.c
+++ b/src/backend/commands/dbcommands.c
@@ -9,13 +9,12 @@
 *
 *
 * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/commands/dbcommands.c,v 1.135 2004/06/10 22:26:18 momjian Exp $
+ *	  $PostgreSQL: pgsql/src/backend/commands/dbcommands.c,v 1.136 2004/06/18 06:13:22 tgl Exp $
 *
 *-------------------------------------------------------------------------
 */
 #include "postgres.h"

-#include <errno.h>
 #include <fcntl.h>
 #include <unistd.h>
 #include <sys/stat.h>
@@ -26,9 +25,12 @@
 #include "catalog/catalog.h"
 #include "catalog/pg_database.h"
 #include "catalog/pg_shadow.h"
+#include "catalog/pg_tablespace.h"
 #include "catalog/indexing.h"
 #include "commands/comment.h"
 #include "commands/dbcommands.h"
+#include "commands/tablespace.h"
+#include "mb/pg_wchar.h"
 #include "miscadmin.h"
 #include "storage/fd.h"
 #include "storage/freespace.h"
@@ -41,32 +43,24 @@
 #include "utils/lsyscache.h"
 #include "utils/syscache.h"

-#include "mb/pg_wchar.h"		/* encoding check */
-

 /* non-export function prototypes */
 static bool get_db_info(const char *name, Oid *dbIdP, int4 *ownerIdP,
 			int *encodingP, bool *dbIsTemplateP, Oid *dbLastSysOidP,
 			TransactionId *dbVacuumXidP, TransactionId *dbFrozenXidP,
-			char *dbpath);
+			Oid *dbTablespace);
 static bool have_createdb_privilege(void);
-static char *resolve_alt_dbpath(const char *dbpath, Oid dboid);
-static bool remove_dbdirs(const char *real_loc, const char *altloc);
+static void remove_dbtablespaces(Oid db_id);
+

 /*
 * CREATE DATABASE
 */
-
 void
 createdb(const CreatedbStmt *stmt)
 {
-	char	   *nominal_loc;
-	char	   *alt_loc;
-	char	   *target_dir;
-	char		src_loc[MAXPGPATH];
-#ifndef WIN32
-	char		buf[2 * MAXPGPATH + 100];
-#endif
+	HeapScanDesc scan;
+	Relation	rel;
 	Oid			src_dboid;
 	AclId		src_owner;
 	int			src_encoding;
@@ -74,7 +68,8 @@ createdb(const CreatedbStmt *stmt)
 	Oid			src_lastsysoid;
 	TransactionId src_vacuumxid;
 	TransactionId src_frozenxid;
-	char		src_dbpath[MAXPGPATH];
+	Oid			src_deftablespace;
+	Oid			dst_deftablespace;
 	Relation	pg_database_rel;
 	HeapTuple	tuple;
 	TupleDesc	pg_database_dsc;
@@ -83,22 +78,35 @@ createdb(const CreatedbStmt *stmt)
 	Oid			dboid;
 	AclId		datdba;
 	ListCell   *option;
+	DefElem	   *dtablespacename = NULL;
 	DefElem    *downer = NULL;
-	DefElem    *dpath = NULL;
 	DefElem    *dtemplate = NULL;
 	DefElem    *dencoding = NULL;
 	char	   *dbname = stmt->dbname;
 	char	   *dbowner = NULL;
-	char	   *dbpath = NULL;
 	char	   *dbtemplate = NULL;
 	int			encoding = -1;
+#ifndef WIN32
+	char		buf[2 * MAXPGPATH + 100];
+#endif
+
+	/* don't call this in a transaction block */
+	PreventTransactionChain((void *) stmt, "CREATE DATABASE");

 	/* Extract options from the statement node tree */
 	foreach(option, stmt->options)
 	{
 		DefElem    *defel = (DefElem *) lfirst(option);

-		if (strcmp(defel->defname, "owner") == 0)
+		if (strcmp(defel->defname, "tablespace") == 0)
+		{
+			if (dtablespacename)
+				ereport(ERROR,
+						(errcode(ERRCODE_SYNTAX_ERROR),
+						 errmsg("conflicting or redundant options")));
+			dtablespacename = defel;
+		}
+		else if (strcmp(defel->defname, "owner") == 0)
 		{
 			if (downer)
 				ereport(ERROR,
@@ -106,14 +114,6 @@ createdb(const CreatedbStmt *stmt)
 						 errmsg("conflicting or redundant options")));
 			downer = defel;
 		}
-		else if (strcmp(defel->defname, "location") == 0)
-		{
-			if (dpath)
-				ereport(ERROR,
-						(errcode(ERRCODE_SYNTAX_ERROR),
-						 errmsg("conflicting or redundant options")));
-			dpath = defel;
-		}
 		else if (strcmp(defel->defname, "template") == 0)
 		{
 			if (dtemplate)
@@ -130,6 +130,13 @@ createdb(const CreatedbStmt *stmt)
 						 errmsg("conflicting or redundant options")));
 			dencoding = defel;
 		}
+		else if (strcmp(defel->defname, "location") == 0)
+		{
+			ereport(WARNING,
+					(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+					 errmsg("LOCATION is not supported anymore"),
+					 errhint("Consider using tablespaces instead.")));
+		}
 		else
 			elog(ERROR, "option \"%s\" not recognized",
 				 defel->defname);
@@ -137,8 +144,6 @@ createdb(const CreatedbStmt *stmt)

 	if (downer && downer->arg)
 		dbowner = strVal(downer->arg);
-	if (dpath && dpath->arg)
-		dbpath = strVal(dpath->arg);
 	if (dtemplate && dtemplate->arg)
 		dbtemplate = strVal(dtemplate->arg);
 	if (dencoding && dencoding->arg)
@@ -195,17 +200,6 @@ createdb(const CreatedbStmt *stmt)
 					 errmsg("must be superuser to create database for another user")));
 	}

-	/* don't call this in a transaction block */
-	PreventTransactionChain((void *) stmt, "CREATE DATABASE");
-
-	/* alternate location requires symlinks */
-#ifndef HAVE_SYMLINK
-	if (dbpath != NULL)
-		ereport(ERROR,
-				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
-		   errmsg("cannot use an alternative location on this platform")));
-#endif
-
 	/*
 	 * Check for db name conflict.	There is a race condition here, since
 	 * another backend could create the same DB name before we commit.
@@ -227,8 +221,7 @@ createdb(const CreatedbStmt *stmt)

 	if (!get_db_info(dbtemplate, &src_dboid, &src_owner, &src_encoding,
 					 &src_istemplate, &src_lastsysoid,
-					 &src_vacuumxid, &src_frozenxid,
-					 src_dbpath))
+					 &src_vacuumxid, &src_frozenxid, &src_deftablespace))
 		ereport(ERROR,
 				(errcode(ERRCODE_UNDEFINED_DATABASE),
 				 errmsg("template database \"%s\" does not exist", dbtemplate)));
@@ -246,14 +239,6 @@ createdb(const CreatedbStmt *stmt)
 							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);
-
 	/*
 	 * The source DB can't have any active backends, except this one
 	 * (exception is to allow CREATE DB while connected to template1).
@@ -276,45 +261,39 @@ createdb(const CreatedbStmt *stmt)
 				(errcode(ERRCODE_WRONG_OBJECT_TYPE),
 				 errmsg("invalid server encoding %d", encoding)));

+	/* Resolve default tablespace for new database */
+	if (dtablespacename && dtablespacename->arg)
+	{
+		char	   *tablespacename;
+        AclResult   aclresult;
+
+		tablespacename = strVal(dtablespacename->arg);
+		dst_deftablespace = get_tablespace_oid(tablespacename);
+		if (!OidIsValid(dst_deftablespace))
+			ereport(ERROR,
+					(errcode(ERRCODE_UNDEFINED_OBJECT),
+					 errmsg("tablespace \"%s\" does not exist",
+							tablespacename)));
+		/* check permissions */
+        aclresult = pg_tablespace_aclcheck(dst_deftablespace, GetUserId(),
+										   ACL_CREATE);
+        if (aclresult != ACLCHECK_OK)
+            aclcheck_error(aclresult, ACL_KIND_TABLESPACE,
+                           tablespacename);
+	}
+	else
+	{
+		/* Use template database's default tablespace */
+		dst_deftablespace = src_deftablespace;
+		/* Note there is no additional permission check in this path */
+	}
+
 	/*
 	 * Preassign OID for pg_database tuple, so that we can compute db
 	 * path.
 	 */
 	dboid = newoid();

-	/*
-	 * Compute nominal location (where we will try to access the
-	 * database), and resolve alternate physical location if one is
-	 * specified.
-	 *
-	 * If an alternate location is specified but is the same as the normal
-	 * path, just drop the alternate-location spec (this seems friendlier
-	 * than erroring out).	We must test this case to avoid creating a
-	 * circular symlink below.
-	 */
-	nominal_loc = GetDatabasePath(dboid);
-	alt_loc = resolve_alt_dbpath(dbpath, dboid);
-
-	if (alt_loc && strcmp(alt_loc, nominal_loc) == 0)
-	{
-		alt_loc = NULL;
-		dbpath = NULL;
-	}
-
-	if (strchr(nominal_loc, '\''))
-		ereport(ERROR,
-				(errcode(ERRCODE_INVALID_NAME),
-				 errmsg("database path may not contain single quotes")));
-	if (alt_loc && strchr(alt_loc, '\''))
-		ereport(ERROR,
-				(errcode(ERRCODE_INVALID_NAME),
-				 errmsg("database path may not contain single quotes")));
-	if (strchr(src_loc, '\''))
-		ereport(ERROR,
-				(errcode(ERRCODE_INVALID_NAME),
-				 errmsg("database path may not contain single quotes")));
-	/* ... otherwise we'd be open to shell exploits below */
-
 	/*
 	 * Force dirty buffers out to disk, to ensure source database is
 	 * up-to-date for the copy.  (We really only need to flush buffers for
@@ -324,74 +303,89 @@ createdb(const CreatedbStmt *stmt)

 	/*
 	 * Close virtual file descriptors so the kernel has more available for
-	 * the mkdir() and system() calls below.
+	 * the system() calls below.
 	 */
 	closeAllVfds();

 	/*
-	 * 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(target_dir, S_IRWXU) != 0)
-		ereport(ERROR,
-				(errcode_for_file_access(),
-				 errmsg("could not create database directory \"%s\": %m",
-						target_dir)));
-	if (rmdir(target_dir) != 0)
-		ereport(ERROR,
-				(errcode_for_file_access(),
-				 errmsg("could not remove temporary directory \"%s\": %m",
-						target_dir)));
-
-	/* Make the symlink, if needed */
-	if (alt_loc)
-	{
-#ifdef HAVE_SYMLINK				/* already throws error above */
-		if (symlink(alt_loc, nominal_loc) != 0)
-#endif
-			ereport(ERROR,
-					(errcode_for_file_access(),
-					 errmsg("could not link file \"%s\" to \"%s\": %m",
-							nominal_loc, alt_loc)));
-	}
-
-	/*
-	 * Copy the template database to the new location
+	 * Iterate through all tablespaces of the template database, and
+	 * copy each one to the new database.
 	 *
-	 * XXX use of cp really makes this code pretty grotty, particularly
-	 * with respect to lack of ability to report errors well.  Someday
-	 * rewrite to do it for ourselves.
+	 * If we are trying to change the default tablespace of the template,
+	 * we require that the template not have any files in the new default
+	 * tablespace.  This avoids the need to merge two subdirectories.
+	 * This could probably be improved later.
 	 */
-#ifndef WIN32
-	/* We might need to use cp -R one day for portability */
-	snprintf(buf, sizeof(buf), "cp -r '%s' '%s'", src_loc, target_dir);
-	if (system(buf) != 0)
+	rel = heap_openr(TableSpaceRelationName, AccessShareLock);
+	scan = heap_beginscan(rel, SnapshotNow, 0, NULL);
+	while ((tuple = heap_getnext(scan, ForwardScanDirection)) != NULL)
 	{
-		if (remove_dbdirs(nominal_loc, alt_loc))
+		Oid		srctablespace = HeapTupleGetOid(tuple);
+		Oid		dsttablespace;
+		char   *srcpath;
+		char   *dstpath;
+		struct stat st;
+
+		/* No need to copy global tablespace */
+		if (srctablespace == GLOBALTABLESPACE_OID)
+			continue;
+
+		srcpath = GetDatabasePath(src_dboid, srctablespace);
+
+		if (stat(srcpath, &st) < 0 || !S_ISDIR(st.st_mode))
+		{
+			/* Assume we can ignore it */
+			pfree(srcpath);
+			continue;
+		}
+
+		if (srctablespace == src_deftablespace)
+			dsttablespace = dst_deftablespace;
+		else
+			dsttablespace = srctablespace;
+
+		dstpath = GetDatabasePath(dboid, dsttablespace);
+
+		if (stat(dstpath, &st) == 0 || errno != ENOENT)
+		{
+			remove_dbtablespaces(dboid);
+			ereport(ERROR,
+					(errmsg("could not initialize database directory"),
+					 errdetail("Directory \"%s\" already exists.", dstpath)));
+		}
+
+#ifndef WIN32
+		/*
+		 * Copy this subdirectory to the new location
+		 *
+		 * XXX use of cp really makes this code pretty grotty, particularly
+		 * with respect to lack of ability to report errors well.  Someday
+		 * rewrite to do it for ourselves.
+		 */
+
+		/* We might need to use cp -R one day for portability */
+		snprintf(buf, sizeof(buf), "cp -r '%s' '%s'",
+				 srcpath, dstpath);
+		if (system(buf) != 0)
+		{
+			remove_dbtablespaces(dboid);
 			ereport(ERROR,
 					(errmsg("could not initialize database directory"),
 					 errdetail("Failing system command was: %s", buf),
 					 errhint("Look in the postmaster's stderr log for more information.")));
-		else
-			ereport(ERROR,
-					(errmsg("could not initialize database directory; delete failed as well"),
-					 errdetail("Failing system command was: %s", buf),
-					 errhint("Look in the postmaster's stderr log for more information.")));
-	}
+		}
 #else	/* WIN32 */
-	if (copydir(src_loc, target_dir) != 0)
-	{
-		/* copydir should already have given details of its troubles */
-		if (remove_dbdirs(nominal_loc, alt_loc))
+		if (copydir(srcpath, dstpath) != 0)
+		{
+			/* copydir should already have given details of its troubles */
+			remove_dbtablespaces(dboid);
 			ereport(ERROR,
 					(errmsg("could not initialize database directory")));
-		else
-			ereport(ERROR,
-					(errmsg("could not initialize database directory; delete failed as well")));
-	}
+		}
 #endif	/* WIN32 */
+	}
+	heap_endscan(scan);
+	heap_close(rel, AccessShareLock);

 	/*
 	 * Now OK to grab exclusive lock on pg_database.
@@ -403,7 +397,7 @@ createdb(const CreatedbStmt *stmt)
 	{
 		/* Don't hold lock while doing recursive remove */
 		heap_close(pg_database_rel, AccessExclusiveLock);
-		remove_dbdirs(nominal_loc, alt_loc);
+		remove_dbtablespaces(dboid);
 		ereport(ERROR,
 				(errcode(ERRCODE_DUPLICATE_DATABASE),
 				 errmsg("database \"%s\" already exists", dbname)));
@@ -427,9 +421,7 @@ createdb(const CreatedbStmt *stmt)
 	new_record[Anum_pg_database_datlastsysoid - 1] = ObjectIdGetDatum(src_lastsysoid);
 	new_record[Anum_pg_database_datvacuumxid - 1] = TransactionIdGetDatum(src_vacuumxid);
 	new_record[Anum_pg_database_datfrozenxid - 1] = TransactionIdGetDatum(src_frozenxid);
-	/* do not set datpath to null, GetRawDatabaseInfo won't cope */
-	new_record[Anum_pg_database_datpath - 1] =
-		DirectFunctionCall1(textin, CStringGetDatum(dbpath ? dbpath : ""));
+	new_record[Anum_pg_database_dattablespace - 1] = ObjectIdGetDatum(dst_deftablespace);

 	/*
 	 * We deliberately set datconfig and datacl to defaults (NULL), rather
@@ -471,14 +463,13 @@ dropdb(const char *dbname)
 	int4		db_owner;
 	bool		db_istemplate;
 	Oid			db_id;
-	char	   *alt_loc;
-	char	   *nominal_loc;
-	char		dbpath[MAXPGPATH];
 	Relation	pgdbrel;
 	SysScanDesc pgdbscan;
 	ScanKeyData key;
 	HeapTuple	tup;

+	PreventTransactionChain((void *) dbname, "DROP DATABASE");
+
 	AssertArg(dbname);

 	if (strcmp(dbname, get_database_name(MyDatabaseId)) == 0)
@@ -486,8 +477,6 @@ dropdb(const char *dbname)
 				(errcode(ERRCODE_OBJECT_IN_USE),
 				 errmsg("cannot drop the currently open database")));

-	PreventTransactionChain((void *) dbname, "DROP DATABASE");
-
 	/*
 	 * Obtain exclusive lock on pg_database.  We need this to ensure that
 	 * no new backend starts up in the target database while we are
@@ -500,7 +489,7 @@ dropdb(const char *dbname)
 	pgdbrel = heap_openr(DatabaseRelationName, AccessExclusiveLock);

 	if (!get_db_info(dbname, &db_id, &db_owner, NULL,
-					 &db_istemplate, NULL, NULL, NULL, dbpath))
+					 &db_istemplate, NULL, NULL, NULL, NULL))
 		ereport(ERROR,
 				(errcode(ERRCODE_UNDEFINED_DATABASE),
 				 errmsg("database \"%s\" does not exist", dbname)));
@@ -519,9 +508,6 @@ dropdb(const char *dbname)
 				(errcode(ERRCODE_WRONG_OBJECT_TYPE),
 				 errmsg("cannot drop a template database")));

-	nominal_loc = GetDatabasePath(db_id);
-	alt_loc = resolve_alt_dbpath(dbpath, db_id);
-
 	/*
 	 * Check for active backends in the target database.
 	 */
@@ -585,9 +571,9 @@ dropdb(const char *dbname)
 	FreeSpaceMapForgetDatabase(db_id);

 	/*
-	 * Remove the database's subdirectory and everything in it.
+	 * Remove all tablespace subdirs belonging to the database.
 	 */
-	remove_dbdirs(nominal_loc, alt_loc);
+	remove_dbtablespaces(db_id);

 	/*
 	 * Force dirty buffers out to disk, so that newly-connecting backends
@@ -831,7 +817,7 @@ static bool
 get_db_info(const char *name, Oid *dbIdP, int4 *ownerIdP,
 			int *encodingP, bool *dbIsTemplateP, Oid *dbLastSysOidP,
 			TransactionId *dbVacuumXidP, TransactionId *dbFrozenXidP,
-			char *dbpath)
+			Oid *dbTablespace)
 {
 	Relation	relation;
 	ScanKeyData scanKey;
@@ -880,28 +866,9 @@ get_db_info(const char *name, Oid *dbIdP, int4 *ownerIdP,
 		/* limit of frozen XIDs */
 		if (dbFrozenXidP)
 			*dbFrozenXidP = dbform->datfrozenxid;
-		/* database path (as registered in pg_database) */
-		if (dbpath)
-		{
-			Datum		datum;
-			bool		isnull;
-
-			datum = heap_getattr(tuple,
-								 Anum_pg_database_datpath,
-								 RelationGetDescr(relation),
-								 &isnull);
-			if (!isnull)
-			{
-				text	   *pathtext = DatumGetTextP(datum);
-				int			pathlen = VARSIZE(pathtext) - VARHDRSZ;
-
-				Assert(pathlen >= 0 && pathlen < MAXPGPATH);
-				strncpy(dbpath, VARDATA(pathtext), pathlen);
-				*(dbpath + pathlen) = '\0';
-			}
-			else
-				strcpy(dbpath, "");
-		}
+		/* default tablespace for this database */
+		if (dbTablespace)
+			*dbTablespace = dbform->dattablespace;
 	}

 	systable_endscan(scan);
@@ -930,105 +897,60 @@ have_createdb_privilege(void)
 	return retval;
 }

-
-static char *
-resolve_alt_dbpath(const char *dbpath, Oid dboid)
+/*
+ * Remove tablespace directories
+ *
+ * We don't know what tablespaces db_id is using, so iterate through all
+ * tablespaces removing <tablespace>/db_id
+ */
+static void
+remove_dbtablespaces(Oid db_id)
 {
-	const char *prefix;
-	char	   *ret;
-	size_t		len;
+	Relation rel;
+	HeapScanDesc scan;
+	HeapTuple tuple;
+	char buf[MAXPGPATH + 100];

-	if (dbpath == NULL || dbpath[0] == '\0')
-		return NULL;
-
-	if (first_dir_separator(dbpath))
+	rel = heap_openr(TableSpaceRelationName, AccessShareLock);
+	scan = heap_beginscan(rel, SnapshotNow, 0, NULL);
+	while ((tuple = heap_getnext(scan, ForwardScanDirection)) != NULL)
 	{
-		if (!is_absolute_path(dbpath))
-			ereport(ERROR,
-					(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
-					 errmsg("relative paths are not allowed as database locations")));
-#ifndef ALLOW_ABSOLUTE_DBPATHS
-		ereport(ERROR,
-				(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
-		errmsg("absolute paths are not allowed as database locations")));
-#endif
-		prefix = dbpath;
-	}
-	else
-	{
-		/* must be environment variable */
-		char	   *var = getenv(dbpath);
+		Oid		dsttablespace = HeapTupleGetOid(tuple);
+		char   *dstpath;
+		struct stat st;

-		if (!var)
-			ereport(ERROR,
-					(errcode(ERRCODE_UNDEFINED_OBJECT),
-			   errmsg("postmaster environment variable \"%s\" not found",
-					  dbpath)));
-		if (!is_absolute_path(var))
-			ereport(ERROR,
-					(errcode(ERRCODE_INVALID_NAME),
-					 errmsg("postmaster environment variable \"%s\" must be absolute path",
-							dbpath)));
-		prefix = var;
-	}
+		/* Don't mess with the global tablespace */
+		if (dsttablespace == GLOBALTABLESPACE_OID)
+			continue;

-	len = strlen(prefix) + 6 + sizeof(Oid) * 8 + 1;
-	if (len >= MAXPGPATH - 100)
-		ereport(ERROR,
-				(errcode(ERRCODE_INVALID_NAME),
-				 errmsg("alternative path is too long")));
+		dstpath = GetDatabasePath(db_id, dsttablespace);

-	ret = palloc(len);
-	snprintf(ret, len, "%s/base/%u", prefix, dboid);
-
-	return ret;
-}
-
-
-static bool
-remove_dbdirs(const char *nominal_loc, const char *alt_loc)
-{
-	const char *target_dir;
-	char		buf[MAXPGPATH + 100];
-	bool		success = true;
-
-	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(nominal_loc) != 0)
+		if (stat(dstpath, &st) < 0 || !S_ISDIR(st.st_mode))
 		{
-			ereport(WARNING,
-					(errcode_for_file_access(),
-					 errmsg("could not remove file \"%s\": %m", nominal_loc)));
-			success = false;
+			/* Assume we can ignore it */
+			pfree(dstpath);
+			continue;
 		}
-	}

 #ifndef WIN32
-	snprintf(buf, sizeof(buf), "rm -rf '%s'", target_dir);
+		snprintf(buf, sizeof(buf), "rm -rf '%s'", dstpath);
 #else
-	snprintf(buf, sizeof(buf), "rmdir /s /q \"%s\"", target_dir);
+		snprintf(buf, sizeof(buf), "rmdir /s /q \"%s\"", dstpath);
 #endif
-
-	if (system(buf) != 0)
-	{
-		ereport(WARNING,
+		if (system(buf) != 0)
+		{
+			ereport(WARNING,
 				(errmsg("could not remove database directory \"%s\"",
-						target_dir),
+						dstpath),
 				 errdetail("Failing system command was: %s", buf),
 				 errhint("Look in the postmaster's stderr log for more information.")));
-		success = false;
+		}
+
+		pfree(dstpath);
 	}

-	return success;
+	heap_endscan(scan);
+	heap_close(rel, AccessShareLock);
 }


@@ -1075,7 +997,7 @@ get_database_oid(const char *dbname)
 /*
 * get_database_name - given a database OID, look up the name
 *
- * Returns InvalidOid if database name not found.
+ * Returns a palloc'd string, or NULL if no such database.
 *
 * This is not actually used in this file, but is exported for use elsewhere.
 */
--- a/src/backend/commands/indexcmds.c
+++ b/src/backend/commands/indexcmds.c
@@ -8,7 +8,7 @@
 *
 *
 * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/commands/indexcmds.c,v 1.121 2004/06/10 17:55:56 tgl Exp $
+ *	  $PostgreSQL: pgsql/src/backend/commands/indexcmds.c,v 1.122 2004/06/18 06:13:23 tgl Exp $
 *
 *-------------------------------------------------------------------------
 */
@@ -27,6 +27,7 @@
 #include "commands/dbcommands.h"
 #include "commands/defrem.h"
 #include "commands/tablecmds.h"
+#include "commands/tablespace.h"
 #include "executor/executor.h"
 #include "mb/pg_wchar.h"
 #include "miscadmin.h"
@@ -64,6 +65,8 @@ static bool relationHasPrimaryKey(Relation rel);
 * 'indexRelationName': the name for the new index, or NULL to indicate
 *		that a nonconflicting default name should be picked.
 * 'accessMethodName': name of the AM to use.
+ * 'tableSpaceName': name of the tablespace to create the index in.
+ *		NULL specifies using the same tablespace as the parent relation.
 * 'attributeList': a list of IndexElem specifying columns and expressions
 *		to index on.
 * 'predicate': the partial-index condition, or NULL if none.
@@ -83,6 +86,7 @@ void
 DefineIndex(RangeVar *heapRelation,
 			char *indexRelationName,
 			char *accessMethodName,
+			char *tableSpaceName,
 			List *attributeList,
 			Expr *predicate,
 			List *rangetable,
@@ -98,6 +102,7 @@ DefineIndex(RangeVar *heapRelation,
 	Oid			accessMethodId;
 	Oid			relationId;
 	Oid			namespaceId;
+	Oid			tablespaceId;
 	Relation	rel;
 	HeapTuple	tuple;
 	Form_pg_am	accessMethodForm;
@@ -151,6 +156,29 @@ DefineIndex(RangeVar *heapRelation,
 						   get_namespace_name(namespaceId));
 	}

+	/* Determine tablespace to use */
+	if (tableSpaceName)
+	{
+		AclResult   aclresult;
+
+		tablespaceId = get_tablespace_oid(tableSpaceName);
+		if (!OidIsValid(tablespaceId))
+			ereport(ERROR,
+					(errcode(ERRCODE_UNDEFINED_OBJECT),
+					 errmsg("tablespace \"%s\" does not exist",
+							tableSpaceName)));
+		/* check permissions */
+		aclresult = pg_tablespace_aclcheck(tablespaceId, GetUserId(),
+										   ACL_CREATE);
+		if (aclresult != ACLCHECK_OK)
+			aclcheck_error(aclresult, ACL_KIND_TABLESPACE,
+						   tableSpaceName);
+	} else {
+		/* Use the parent rel's tablespace */
+		tablespaceId = get_rel_tablespace(relationId);
+		/* Note there is no additional permission check in this path */
+	}
+
 	/*
 	 * Select name for index if caller didn't specify
 	 */
@@ -335,7 +363,7 @@ DefineIndex(RangeVar *heapRelation,
 						indexRelationName, RelationGetRelationName(rel))));

 	index_create(relationId, indexRelationName,
-				 indexInfo, accessMethodId, classObjectId,
+				 indexInfo, accessMethodId, tablespaceId, classObjectId,
 				 primary, isconstraint,
 				 allowSystemTableMods, skip_build);

--- a/src/backend/commands/schemacmds.c
+++ b/src/backend/commands/schemacmds.c
@@ -8,7 +8,7 @@
 *
 *
 * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/commands/schemacmds.c,v 1.18 2004/05/26 04:41:11 neilc Exp $
+ *	  $PostgreSQL: pgsql/src/backend/commands/schemacmds.c,v 1.19 2004/06/18 06:13:23 tgl Exp $
 *
 *-------------------------------------------------------------------------
 */
@@ -23,6 +23,7 @@
 #include "catalog/pg_namespace.h"
 #include "commands/dbcommands.h"
 #include "commands/schemacmds.h"
+#include "commands/tablespace.h"
 #include "miscadmin.h"
 #include "parser/analyze.h"
 #include "tcop/utility.h"
@@ -41,6 +42,7 @@ CreateSchemaCommand(CreateSchemaStmt *stmt)
 	const char *schemaName = stmt->schemaname;
 	const char *authId = stmt->authid;
 	Oid			namespaceId;
+	Oid			tablespaceId;
 	List	   *parsetree_list;
 	ListCell   *parsetree_item;
 	const char *owner_name;
@@ -100,8 +102,33 @@ CreateSchemaCommand(CreateSchemaStmt *stmt)
 				 errmsg("unacceptable schema name \"%s\"", schemaName),
 		errdetail("The prefix \"pg_\" is reserved for system schemas.")));

+	/*
+	 * Select default tablespace for schema.  If not given, use zero
+	 * which implies the database's default tablespace.
+	 */
+	if (stmt->tablespacename)
+	{
+		AclResult   aclresult;
+
+		tablespaceId = get_tablespace_oid(stmt->tablespacename);
+		if (!OidIsValid(tablespaceId))
+			ereport(ERROR,
+					(errcode(ERRCODE_UNDEFINED_OBJECT),
+					 errmsg("tablespace \"%s\" does not exist",
+							stmt->tablespacename)));
+		/* check permissions */
+		aclresult = pg_tablespace_aclcheck(tablespaceId, GetUserId(),
+										   ACL_CREATE);
+		if (aclresult != ACLCHECK_OK)
+			aclcheck_error(aclresult, ACL_KIND_TABLESPACE,
+						   stmt->tablespacename);
+	} else {
+		tablespaceId = InvalidOid;
+		/* note there is no permission check in this path */
+	}
+
 	/* Create the schema's namespace */
-	namespaceId = NamespaceCreate(schemaName, owner_userid);
+	namespaceId = NamespaceCreate(schemaName, owner_userid, tablespaceId);

 	/* Advance cmd counter to make the namespace visible */
 	CommandCounterIncrement();
--- a/src/backend/commands/sequence.c
+++ b/src/backend/commands/sequence.c
@@ -8,7 +8,7 @@
 *
 *
 * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/commands/sequence.c,v 1.111 2004/05/26 04:41:11 neilc Exp $
+ *	  $PostgreSQL: pgsql/src/backend/commands/sequence.c,v 1.112 2004/06/18 06:13:23 tgl Exp $
 *
 *-------------------------------------------------------------------------
 */
@@ -180,6 +180,7 @@ DefineSequence(CreateSeqStmt *seq)
 	stmt->constraints = NIL;
 	stmt->hasoids = MUST_NOT_HAVE_OIDS;
 	stmt->oncommit = ONCOMMIT_NOOP;
+	stmt->tablespacename = seq->tablespacename;

 	seqoid = DefineRelation(stmt, RELKIND_SEQUENCE);

@@ -1071,8 +1072,8 @@ seq_redo(XLogRecPtr lsn, XLogRecord *record)

 	buffer = XLogReadBuffer(true, reln, 0);
 	if (!BufferIsValid(buffer))
-		elog(PANIC, "seq_redo: can't read block of %u/%u",
-			 xlrec->node.tblNode, xlrec->node.relNode);
+		elog(PANIC, "seq_redo: can't read block 0 of rel %u/%u/%u",
+			 xlrec->node.spcNode, xlrec->node.dbNode, xlrec->node.relNode);

 	page = (Page) BufferGetPage(buffer);

@@ -1114,6 +1115,6 @@ seq_desc(char *buf, uint8 xl_info, char *rec)
 		return;
 	}

-	sprintf(buf + strlen(buf), "node %u/%u",
-			xlrec->node.tblNode, xlrec->node.relNode);
+	sprintf(buf + strlen(buf), "rel %u/%u/%u",
+			xlrec->node.spcNode, xlrec->node.dbNode, xlrec->node.relNode);
 }
--- a/src/backend/commands/tablecmds.c
+++ b/src/backend/commands/tablecmds.c
@@ -8,7 +8,7 @@
 *
 *
 * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/commands/tablecmds.c,v 1.115 2004/06/10 18:34:45 tgl Exp $
+ *	  $PostgreSQL: pgsql/src/backend/commands/tablecmds.c,v 1.116 2004/06/18 06:13:23 tgl Exp $
 *
 *-------------------------------------------------------------------------
 */
@@ -33,6 +33,7 @@
 #include "commands/cluster.h"
 #include "commands/defrem.h"
 #include "commands/tablecmds.h"
+#include "commands/tablespace.h"
 #include "commands/trigger.h"
 #include "executor/executor.h"
 #include "lib/stringinfo.h"
@@ -258,6 +259,7 @@ DefineRelation(CreateStmt *stmt, char relkind)
 	Oid			namespaceId;
 	List	   *schema = stmt->tableElts;
 	Oid			relationId;
+	Oid			tablespaceId;
 	Relation	rel;
 	TupleDesc	descriptor;
 	List	   *inheritOids;
@@ -301,6 +303,31 @@ DefineRelation(CreateStmt *stmt, char relkind)
 						   get_namespace_name(namespaceId));
 	}

+	/*
+	 * Select tablespace to use.  If not specified, use containing schema's
+	 * default tablespace (which may in turn default to database's default).
+	 */
+	if (stmt->tablespacename)
+	{
+		AclResult   aclresult;
+
+		tablespaceId = get_tablespace_oid(stmt->tablespacename);
+		if (!OidIsValid(tablespaceId))
+			ereport(ERROR,
+					(errcode(ERRCODE_UNDEFINED_OBJECT),
+					 errmsg("tablespace \"%s\" does not exist",
+							stmt->tablespacename)));
+		/* check permissions */
+		aclresult = pg_tablespace_aclcheck(tablespaceId, GetUserId(),
+										   ACL_CREATE);
+		if (aclresult != ACLCHECK_OK)
+			aclcheck_error(aclresult, ACL_KIND_TABLESPACE,
+						   stmt->tablespacename);
+	} else {
+		tablespaceId = get_namespace_tablespace(namespaceId);
+		/* note no permission check on tablespace in this case */
+	}
+
 	/*
 	 * Look up inheritance ancestors and generate relation schema,
 	 * including inherited attributes.
@@ -379,6 +406,7 @@ DefineRelation(CreateStmt *stmt, char relkind)

 	relationId = heap_create_with_catalog(relname,
 										  namespaceId,
+										  tablespaceId,
 										  descriptor,
 										  relkind,
 										  false,
@@ -1770,7 +1798,7 @@ ATController(Relation rel, List *cmds, bool recurse)
 /*
 * ATPrepCmd
 *
- * Traffic cop for ALTER TABLE Phase 1 operations, including simple 
+ * Traffic cop for ALTER TABLE Phase 1 operations, including simple
 * recursion and permission checks.
 *
 * Caller must have acquired AccessExclusiveLock on relation already.
@@ -2679,7 +2707,7 @@ find_composite_type_dependencies(Oid typeOid, const char *origTblName)
 }


-/* 
+/*
 * ALTER TABLE ADD COLUMN
 *
 * Adds an additional attribute to a relation making the assumption that
@@ -3521,6 +3549,7 @@ ATExecAddIndex(AlteredTableInfo *tab, Relation rel,
 	DefineIndex(stmt->relation,		/* relation */
 				stmt->idxname,		/* index name */
 				stmt->accessMethod, /* am name */
+				stmt->tableSpace,
 				stmt->indexParams,	/* parameters */
 				(Expr *) stmt->whereClause,
 				stmt->rangetable,
@@ -3566,7 +3595,7 @@ ATExecAddConstraint(AlteredTableInfo *tab, Relation rel, Node *newConstraint)
 														list_make1(constr));
 					/* Add each constraint to Phase 3's queue */
 					foreach(lcon, newcons)
-					{ 
+					{
 						CookedConstraint *ccon = (CookedConstraint *) lfirst(lcon);
 						NewConstraint *newcon;

@@ -3643,7 +3672,7 @@ ATAddForeignKeyConstraint(AlteredTableInfo *tab, Relation rel,
 	int16		fkattnum[INDEX_MAX_KEYS];
 	Oid			pktypoid[INDEX_MAX_KEYS];
 	Oid			fktypoid[INDEX_MAX_KEYS];
-	Oid         opclasses[INDEX_MAX_KEYS];
+	Oid			opclasses[INDEX_MAX_KEYS];
 	int			i;
 	int			numfks,
 				numpks;
@@ -3791,7 +3820,7 @@ ATAddForeignKeyConstraint(AlteredTableInfo *tab, Relation rel,
 		 * will be incurred to check FK validity.
 		 */
 		if (!op_in_opclass(oprid(o), opclasses[i]))
-			ereport(WARNING, 
+			ereport(WARNING,
 					(errmsg("foreign key constraint \"%s\" "
 							"will require costly sequential scans",
 							fkconstraint->constr_name),
@@ -4565,7 +4594,7 @@ ATPrepAlterColumnType(List **wqueue,
 	/*
 	 * Add a work queue item to make ATRewriteTable update the column
 	 * contents.
-	 */ 
+	 */
 	newval = (NewColumnValue *) palloc0(sizeof(NewColumnValue));
 	newval->attnum = attnum;
 	newval->expr = (Expr *) transform;
@@ -5272,6 +5301,7 @@ AlterTableCreateToastTable(Oid relOid, bool silent)
 	 */
 	toast_relid = heap_create_with_catalog(toast_relname,
 										   PG_TOAST_NAMESPACE,
+										   rel->rd_rel->reltablespace,
 										   tupdesc,
 										   RELKIND_TOASTVALUE,
 										   shared_relation,
@@ -5309,7 +5339,9 @@ AlterTableCreateToastTable(Oid relOid, bool silent)
 	classObjectId[1] = INT4_BTREE_OPS_OID;

 	toast_idxid = index_create(toast_relid, toast_idxname, indexInfo,
-							   BTREE_AM_OID, classObjectId,
+							   BTREE_AM_OID,
+							   rel->rd_rel->reltablespace,
+							   classObjectId,
 							   true, false, true, false);

 	/*
--- a/src/backend/commands/tablespace.c
+++ b/src/backend/commands/tablespace.c
@@ -0,0 +1,660 @@
+/*-------------------------------------------------------------------------
+ *
+ * tablespace.c
+ *	  Commands to manipulate table spaces
+ *
+ *
+ * Tablespaces in PostgreSQL are designed to allow users to determine
+ * where the data file(s) for a given database object reside on the file
+ * system.
+ *
+ * A tablespace represents a directory on the file system. At tablespace
+ * creation time, the directory must be empty. To simplify things and
+ * remove the possibility of having file name conflicts, we isolate
+ * files within a tablespace into database-specific subdirectories.
+ *
+ * To support file access via the information given in RelFileNode, we
+ * maintain a symbolic-link map in $PGDATA/pg_tablespaces. The symlinks are
+ * named by tablespace OIDs and point to the actual tablespace directories.
+ * Thus the full path to an arbitrary file is
+ *			$PGDATA/pg_tablespaces/spcoid/dboid/relfilenode
+ *
+ * There are two tablespaces created at initdb time: global (for shared
+ * tables) and default (for everything else).  For backwards compatibility
+ * and to remain functional on platforms without symlinks, these tablespaces
+ * are accessed specially: they are respectively
+ *			$PGDATA/global/relfilenode
+ *			$PGDATA/base/dboid/relfilenode
+ *
+ * The implementation is designed to be backwards compatible. For this reason
+ * (and also as a feature unto itself) when a user creates an object without
+ * specifying a tablespace, we look at the object's parent and place
+ * the object in the parent's tablespace. The hierarchy is as follows:
+ *			database > schema > table > index
+ *
+ * To allow CREATE DATABASE to give a new database a default tablespace
+ * that's different from the template database's default, we make the
+ * provision that a zero in pg_class.reltablespace means the database's
+ * default tablespace.  Without this, CREATE DATABASE would have to go in
+ * and munge the system catalogs of the new database.  This special meaning
+ * of zero also applies in pg_namespace.nsptablespace.
+ *
+ *
+ * Portions Copyright (c) 1996-2004, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ *
+ * IDENTIFICATION
+ *	  $PostgreSQL: pgsql/src/backend/commands/tablespace.c,v 1.1 2004/06/18 06:13:23 tgl Exp $
+ *
+ *-------------------------------------------------------------------------
+ */
+#include "postgres.h"
+
+#include <unistd.h>
+#include <dirent.h>
+#include <sys/types.h>
+#include <sys/stat.h>
+
+#include "access/heapam.h"
+#include "catalog/catalog.h"
+#include "catalog/catname.h"
+#include "catalog/indexing.h"
+#include "catalog/pg_namespace.h"
+#include "catalog/pg_tablespace.h"
+#include "commands/tablespace.h"
+#include "miscadmin.h"
+#include "storage/fd.h"
+#include "storage/smgr.h"
+#include "utils/acl.h"
+#include "utils/builtins.h"
+#include "utils/fmgroids.h"
+#include "utils/lsyscache.h"
+#include "utils/syscache.h"
+
+
+static void set_short_version(const char *path);
+static bool directory_is_empty(const char *path);
+
+
+/*
+ * Each database using a table space is isolated into its own name space
+ * by a subdirectory named for the database OID.  On first creation of an
+ * object in the tablespace, create the subdirectory.  If the subdirectory
+ * already exists, just fall through quietly.
+ *
+ * If tablespaces are not supported, this is just a no-op; CREATE DATABASE
+ * is expected to create the default subdirectory for the database.
+ */
+void
+TablespaceCreateDbspace(Oid spcNode, Oid dbNode)
+{
+#ifdef HAVE_SYMLINK
+	struct stat st;
+	char		*dir;
+
+	/*
+	 * The global tablespace doesn't have per-database subdirectories,
+	 * so nothing to do for it.
+	 */
+	if (spcNode == GLOBALTABLESPACE_OID)
+		return;
+
+	Assert(OidIsValid(spcNode));
+	Assert(OidIsValid(dbNode));
+
+	dir = GetDatabasePath(dbNode, spcNode);
+
+	if (stat(dir, &st) < 0)
+	{
+		if (errno == ENOENT)
+		{
+			/*
+			 * Acquire ExclusiveLock on pg_tablespace to ensure that no
+			 * DROP TABLESPACE or TablespaceCreateDbspace is running
+			 * concurrently.  Simple reads from pg_tablespace are OK.
+			 */
+			Relation rel;
+
+			rel = heap_openr(TableSpaceRelationName, ExclusiveLock);
+
+			/*
+			 * Recheck to see if someone created the directory while
+			 * we were waiting for lock.
+			 */
+			if (stat(dir, &st) == 0 && S_ISDIR(st.st_mode))
+			{
+				/* need not do anything */
+			}
+			else
+			{
+				/* OK, go for it */
+				if (mkdir(dir, S_IRWXU) < 0)
+					ereport(ERROR,
+							(errcode_for_file_access(),
+							 errmsg("could not create directory \"%s\": %m",
+									dir)));
+			}
+
+			/* OK to drop the exclusive lock */
+			heap_close(rel, ExclusiveLock);
+		}
+		else
+		{
+			ereport(ERROR,
+					(errcode_for_file_access(),
+					 errmsg("could not stat directory \"%s\": %m", dir)));
+		}
+	}
+	else
+	{
+		/* be paranoid */
+		if (!S_ISDIR(st.st_mode))
+			ereport(ERROR,
+					(errcode(ERRCODE_WRONG_OBJECT_TYPE),
+					 errmsg("\"%s\" exists but is not a directory",
+							dir)));
+	}
+
+	pfree(dir);
+#endif /* HAVE_SYMLINK */
+}
+
+/*
+ * Create a table space
+ *
+ * Only superusers can create a tablespace. This seems a reasonable restriction
+ * since we're determining the system layout and, anyway, we probably have
+ * root if we're doing this kind of activity
+ */
+void
+CreateTableSpace(CreateTableSpaceStmt *stmt)
+{
+#ifdef HAVE_SYMLINK
+	Relation 	rel;
+	Datum 		values[Natts_pg_tablespace];
+	char		nulls[Natts_pg_tablespace];
+	HeapTuple	tuple;
+	Oid			tablespaceoid;
+	char		*location;
+	char		*linkloc;
+	AclId		ownerid;
+
+	/* validate */
+
+	/* don't call this in a transaction block */
+	PreventTransactionChain((void *) stmt, "CREATE TABLESPACE");
+
+	/* Must be super user */
+	if (!superuser())
+		ereport(ERROR,
+			(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
+			 errmsg("permission denied to create tablespace \"%s\"",
+					stmt->tablespacename),
+			 errhint("Must be superuser to create a tablespace.")));
+
+	/* However, the eventual owner of the tablespace need not be */
+	if (stmt->owner)
+	{
+		/* No need to check result, get_usesysid() does that */
+		ownerid = get_usesysid(stmt->owner);
+	}
+	else
+		ownerid = GetUserId();
+
+	/* Unix-ify the offered path, and strip any trailing slashes */
+	location = pstrdup(stmt->location);
+	canonicalize_path(location);
+
+	/* disallow quotes, else CREATE DATABASE would be at risk */
+	if (strchr(location, '\''))
+		ereport(ERROR,
+				(errcode(ERRCODE_INVALID_NAME),
+				 errmsg("tablespace location may not contain single quotes")));
+
+	/*
+	 * Allowing relative paths seems risky
+	 *
+	 * this also helps us ensure that location is not empty or whitespace
+	 */
+	if (!is_absolute_path(location))
+		ereport(ERROR,
+				(errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
+				 errmsg("tablespace location must be an absolute path")));
+
+	/*
+	 * Check that location isn't too long. Remember that we're going to append
+	 * '/<dboid>/<relid>.<nnn>'  (XXX but do we ever form the whole path
+	 * explicitly?  This may be overly conservative.)
+	 */
+	if (strlen(location) >= (MAXPGPATH - 1 - 10 - 1 - 10 - 1 - 10))
+		ereport(ERROR,
+				(errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
+				 errmsg("tablespace location \"%s\" is too long",
+						location)));
+
+	/*
+	 * Check that there is no other tablespace by this name.  (The
+	 * unique index would catch this anyway, but might as well give 
+	 * a friendlier message.)
+	 */
+	if (OidIsValid(get_tablespace_oid(stmt->tablespacename)))
+		ereport(ERROR,
+				(errcode(ERRCODE_DUPLICATE_OBJECT),
+				 errmsg("tablespace \"%s\" already exists",
+						stmt->tablespacename)));
+
+	/*
+	 * Insert tuple into pg_tablespace.  The purpose of doing this first
+	 * is to lock the proposed tablename against other would-be creators.
+	 * The insertion will roll back if we find problems below.
+	 */
+	rel = heap_openr(TableSpaceRelationName, RowExclusiveLock);
+
+	MemSet(nulls, ' ', Natts_pg_tablespace);
+
+	values[Anum_pg_tablespace_spcname - 1] =
+		DirectFunctionCall1(namein, CStringGetDatum(stmt->tablespacename));
+	values[Anum_pg_tablespace_spcowner - 1] =
+		Int32GetDatum(ownerid);
+	values[Anum_pg_tablespace_spclocation - 1] =
+		DirectFunctionCall1(textin, CStringGetDatum(location));
+	nulls[Anum_pg_tablespace_spcacl - 1] = 'n';
+
+	tuple = heap_formtuple(rel->rd_att, values, nulls);
+
+	tablespaceoid = newoid();
+
+	HeapTupleSetOid(tuple, tablespaceoid);
+
+	simple_heap_insert(rel, tuple);
+
+	CatalogUpdateIndexes(rel, tuple);
+
+	heap_freetuple(tuple);
+
+	/*
+	 * Attempt to coerce target directory to safe permissions.  If this
+	 * fails, it doesn't exist or has the wrong owner.
+	 */
+	if (chmod(location, 0700) != 0)
+		ereport(ERROR,
+				(errcode_for_file_access(),
+				 errmsg("could not set permissions on directory \"%s\": %m",
+						location)));
+
+	/*
+	 * Check the target directory is empty.
+	 */
+	if (!directory_is_empty(location))
+		ereport(ERROR,
+				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+				 errmsg("directory \"%s\" is not empty",
+						location)));
+
+	/*
+	 * Create the PG_VERSION file in the target directory.  This has several
+	 * purposes: to make sure we can write in the directory, to prevent
+	 * someone from creating another tablespace pointing at the same
+	 * directory (the emptiness check above will fail), and to label
+	 * tablespace directories by PG version.
+	 */
+	set_short_version(location);
+
+	/*
+	 * All seems well, create the symlink
+	 */
+	linkloc = (char *) palloc(strlen(DataDir) + 16 + 10 + 1);
+	sprintf(linkloc, "%s/pg_tablespaces/%u", DataDir, tablespaceoid);
+
+	if (symlink(location, linkloc) < 0)
+		ereport(ERROR,
+				(errcode_for_file_access(),
+				 errmsg("could not create symbolic link \"%s\": %m",
+						linkloc)));
+
+	pfree(linkloc);
+	pfree(location);
+
+	heap_close(rel, RowExclusiveLock);
+
+#else /* !HAVE_SYMLINK */
+	ereport(ERROR,
+			(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+			 errmsg("tablespaces are not supported on this platform")));
+#endif /* HAVE_SYMLINK */
+}
+
+/*
+ * Drop a table space
+ *
+ * Be careful to check that the tablespace is empty.
+ */
+void
+DropTableSpace(DropTableSpaceStmt *stmt)
+{
+#ifdef HAVE_SYMLINK
+	char 		   *tablespacename = stmt->tablespacename;
+	HeapScanDesc 	scandesc;
+	Relation 		rel;
+	HeapTuple   	tuple;
+	ScanKeyData 	entry[1];
+	char		   *location;
+	Oid				tablespaceoid;
+	DIR			*dirdesc;
+	struct dirent *de;
+	char *subfile;
+
+	/* don't call this in a transaction block */
+	PreventTransactionChain((void *) stmt, "DROP TABLESPACE");
+
+	/*
+	 * Acquire ExclusiveLock on pg_tablespace to ensure that no one else
+	 * is trying to do DROP TABLESPACE or TablespaceCreateDbspace concurrently.
+	 */
+	rel = heap_openr(TableSpaceRelationName, ExclusiveLock);
+
+	/*
+	 * Find the target tuple
+	 */
+	ScanKeyInit(&entry[0],
+				Anum_pg_tablespace_spcname,
+				BTEqualStrategyNumber, F_NAMEEQ,
+				CStringGetDatum(tablespacename));
+	scandesc = heap_beginscan(rel, SnapshotNow, 1, entry);
+	tuple = heap_getnext(scandesc, ForwardScanDirection);
+
+	if (!HeapTupleIsValid(tuple))
+		ereport(ERROR,
+				(errcode(ERRCODE_UNDEFINED_OBJECT),
+				 errmsg("tablespace \"%s\" does not exist",
+						tablespacename)));
+
+	tablespaceoid = HeapTupleGetOid(tuple);
+
+	/* Must be superuser or owner */
+	if (GetUserId() != ((Form_pg_tablespace) GETSTRUCT(tuple))->spcowner &&
+		!superuser())
+		aclcheck_error(ACLCHECK_NOT_OWNER, ACL_KIND_TABLESPACE,
+					   tablespacename);
+
+	/* Disallow drop of the standard tablespaces, even by superuser */
+	if (tablespaceoid == GLOBALTABLESPACE_OID ||
+		tablespaceoid == DEFAULTTABLESPACE_OID)
+		aclcheck_error(ACLCHECK_NO_PRIV, ACL_KIND_TABLESPACE,
+					   tablespacename);
+
+	location = (char *) palloc(strlen(DataDir) + 16 + 10 + 1);
+	sprintf(location, "%s/pg_tablespaces/%u", DataDir, tablespaceoid);
+
+	/*
+	 * Check if the tablespace still contains any files.  We try to rmdir
+	 * each per-database directory we find in it.  rmdir failure implies
+	 * there are still files in that subdirectory, so give up.  (We do not
+	 * have to worry about undoing any already completed rmdirs, since
+	 * the next attempt to use the tablespace from that database will simply
+	 * recreate the subdirectory via TablespaceCreateDbspace.)
+	 *
+	 * Since we hold exclusive lock, no one else should be creating any
+	 * fresh subdirectories in parallel.  It is possible that new files
+	 * are being created within subdirectories, though, so the rmdir
+	 * call could fail.  Worst consequence is a less friendly error message.
+	 */
+	dirdesc = AllocateDir(location);
+	if (dirdesc == NULL)
+		ereport(ERROR,
+				(errcode_for_file_access(),
+				 errmsg("could not open directory \"%s\": %m",
+						location)));
+
+	errno = 0;
+	while ((de = readdir(dirdesc)) != NULL)
+	{
+		/* Note we ignore PG_VERSION for the nonce */
+		if (strcmp(de->d_name, ".") == 0 ||
+			strcmp(de->d_name, "..") == 0 ||
+			strcmp(de->d_name, "PG_VERSION") == 0)
+		{
+			errno = 0;
+			continue;
+		}
+
+		subfile = palloc(strlen(location) + 1 + strlen(de->d_name) + 1);
+		sprintf(subfile, "%s/%s", location, de->d_name);
+
+		/* This check is just to deliver a friendlier error message */
+		if (!directory_is_empty(subfile))
+			ereport(ERROR,
+					(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+					 errmsg("tablespace \"%s\" is not empty",
+							tablespacename)));
+
+		/* Do the real deed */
+		if (rmdir(subfile) < 0)
+			ereport(ERROR,
+					(errcode_for_file_access(),
+					 errmsg("could not delete directory \"%s\": %m",
+							subfile)));
+
+		pfree(subfile);
+	}
+#ifdef WIN32
+	/* This fix is in mingw cvs (runtime/mingwex/dirent.c rev 1.4), but
+	   not in released version */
+	if (GetLastError() == ERROR_NO_MORE_FILES)
+		errno = 0;
+#endif
+	if (errno)
+		ereport(ERROR,
+				(errcode_for_file_access(),
+				 errmsg("could not read directory \"%s\": %m",
+						location)));
+	FreeDir(dirdesc);
+
+	/*
+	 * Okay, try to unlink PG_VERSION and then remove the symlink.
+	 */
+	subfile = palloc(strlen(location) + 11 + 1);
+	sprintf(subfile, "%s/PG_VERSION", location);
+
+	if (unlink(subfile) < 0)
+		ereport(ERROR,
+				(errcode_for_file_access(),
+				 errmsg("could not unlink file \"%s\": %m",
+						subfile)));
+
+	if (unlink(location) < 0)
+		ereport(ERROR,
+				(errcode_for_file_access(),
+				 errmsg("could not unlink symbolic link \"%s\": %m",
+						location)));
+
+	pfree(subfile);
+	pfree(location);
+
+	/*
+	 * We have successfully destroyed the infrastructure ... there is
+	 * now no way to roll back the DROP ... so proceed to remove the
+	 * pg_tablespace tuple.
+	 */
+	simple_heap_delete(rel, &tuple->t_self);
+
+	heap_endscan(scandesc);
+
+	heap_close(rel, ExclusiveLock);
+
+#else /* !HAVE_SYMLINK */
+	ereport(ERROR,
+			(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+			 errmsg("tablespaces are not supported on this platform")));
+#endif /* HAVE_SYMLINK */
+}
+
+
+/*
+ * write out the PG_VERSION file in the specified directory
+ */
+static void
+set_short_version(const char *path)
+{
+	char	   *short_version;
+	bool		gotdot = false;
+	int			end;
+	char	   *fullname;
+	FILE	   *version_file;
+
+	/* Construct short version string (should match initdb.c) */
+	short_version = pstrdup(PG_VERSION);
+
+	for (end = 0; short_version[end] != '\0'; end++)
+	{
+		if (short_version[end] == '.')
+		{
+			Assert(end != 0);
+			if (gotdot)
+				break;
+			else
+				gotdot = true;
+		}
+		else if (short_version[end] < '0' || short_version[end] > '9')
+		{
+			/* gone past digits and dots */
+			break;
+		}
+	}
+	Assert(end > 0 && short_version[end - 1] != '.' && gotdot);
+	short_version[end] = '\0';
+
+	/* Now write the file */
+	fullname = palloc(strlen(path) + 11 + 1);
+	sprintf(fullname, "%s/PG_VERSION", path);
+	version_file = AllocateFile(fullname, PG_BINARY_W);
+	if (version_file == NULL)
+		ereport(ERROR,
+				(errcode_for_file_access(),
+				 errmsg("could not write to file \"%s\": %m",
+						fullname)));
+	fprintf(version_file, "%s\n", short_version);
+	if (FreeFile(version_file))
+		ereport(ERROR,
+				(errcode_for_file_access(),
+				 errmsg("could not write to file \"%s\": %m",
+						fullname)));
+
+	pfree(fullname);
+	pfree(short_version);
+}
+
+/*
+ * Check if a directory is empty.
+ */
+static bool
+directory_is_empty(const char *path)
+{
+	DIR			*dirdesc;
+	struct dirent *de;
+
+	dirdesc = AllocateDir(path);
+	if (dirdesc == NULL)
+		ereport(ERROR,
+				(errcode_for_file_access(),
+				 errmsg("could not open directory \"%s\": %m",
+						path)));
+
+	errno = 0;
+	while ((de = readdir(dirdesc)) != NULL)
+	{
+		if (strcmp(de->d_name, ".") == 0 ||
+			strcmp(de->d_name, "..") == 0)
+		{
+			errno = 0;
+			continue;
+		}
+		FreeDir(dirdesc);
+		return false;
+	}
+#ifdef WIN32
+	/* This fix is in mingw cvs (runtime/mingwex/dirent.c rev 1.4), but
+	   not in released version */
+	if (GetLastError() == ERROR_NO_MORE_FILES)
+		errno = 0;
+#endif
+	if (errno)
+		ereport(ERROR,
+				(errcode_for_file_access(),
+				 errmsg("could not read directory \"%s\": %m",
+						path)));
+	FreeDir(dirdesc);
+	return true;
+}
+
+/*
+ * get_tablespace_oid - given a tablespace name, look up the OID
+ *
+ * Returns InvalidOid if tablespace name not found.
+ */
+Oid
+get_tablespace_oid(const char *tablespacename)
+{
+	Oid result;
+	Relation rel;
+	HeapScanDesc scandesc;
+	HeapTuple	tuple;
+	ScanKeyData	entry[1];
+
+	/* Search pg_tablespace */
+	rel = heap_openr(TableSpaceRelationName, AccessShareLock);
+
+	ScanKeyInit(&entry[0],
+				Anum_pg_tablespace_spcname,
+				BTEqualStrategyNumber, F_NAMEEQ,
+				CStringGetDatum(tablespacename));
+	scandesc = heap_beginscan(rel, SnapshotNow, 1, entry);
+	tuple = heap_getnext(scandesc, ForwardScanDirection);
+
+	if (HeapTupleIsValid(tuple))
+		result = HeapTupleGetOid(tuple);
+	else
+		result = InvalidOid;
+
+    heap_endscan(scandesc);
+    heap_close(rel, AccessShareLock);
+
+	return result;
+}
+
+/*
+ * get_tablespace_name - given a tablespace OID, look up the name
+ *
+ * Returns a palloc'd string, or NULL if no such tablespace.
+ */
+char *
+get_tablespace_name(Oid spc_oid)
+{
+	char *result;
+	Relation rel;
+	HeapScanDesc scandesc;
+	HeapTuple	tuple;
+	ScanKeyData	entry[1];
+
+	/* Search pg_tablespace */
+	rel = heap_openr(TableSpaceRelationName, AccessShareLock);
+
+	ScanKeyInit(&entry[0],
+				ObjectIdAttributeNumber,
+				BTEqualStrategyNumber, F_OIDEQ,
+				ObjectIdGetDatum(spc_oid));
+	scandesc = heap_beginscan(rel, SnapshotNow, 1, entry);
+	tuple = heap_getnext(scandesc, ForwardScanDirection);
+
+	/* We assume that there can be at most one matching tuple */
+	if (HeapTupleIsValid(tuple))
+		result = pstrdup(NameStr(((Form_pg_tablespace) GETSTRUCT(tuple))->spcname));
+	else
+		result = NULL;
+
+    heap_endscan(scandesc);
+    heap_close(rel, AccessShareLock);
+
+	return result;
+}
--- a/src/backend/commands/typecmds.c
+++ b/src/backend/commands/typecmds.c
@@ -8,7 +8,7 @@
 *
 *
 * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/commands/typecmds.c,v 1.59 2004/06/10 17:55:56 tgl Exp $
+ *	  $PostgreSQL: pgsql/src/backend/commands/typecmds.c,v 1.60 2004/06/18 06:13:23 tgl Exp $
 *
 * DESCRIPTION
 *	  The "DefineFoo" routines take the parse tree and pick out the
@@ -1110,8 +1110,8 @@ DefineCompositeType(const RangeVar *typevar, List *coldeflist)
 			 errmsg("composite type must have at least one attribute")));

 	/*
-	 * now create the parameters for keys/inheritance etc. All of them are
-	 * nil...
+	 * now set the parameters for keys/inheritance etc. All of these
+	 * are uninteresting for composite types...
 	 */
 	createStmt->relation = (RangeVar *) typevar;
 	createStmt->tableElts = coldeflist;
@@ -1119,6 +1119,7 @@ DefineCompositeType(const RangeVar *typevar, List *coldeflist)
 	createStmt->constraints = NIL;
 	createStmt->hasoids = MUST_NOT_HAVE_OIDS;
 	createStmt->oncommit = ONCOMMIT_NOOP;
+	createStmt->tablespacename = NULL;

 	/*
 	 * finally create the relation...
--- a/src/backend/commands/view.c
+++ b/src/backend/commands/view.c
@@ -8,7 +8,7 @@
 *
 *
 * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/commands/view.c,v 1.82 2004/05/26 04:41:13 neilc Exp $
+ *	  $PostgreSQL: pgsql/src/backend/commands/view.c,v 1.83 2004/06/18 06:13:23 tgl Exp $
 *
 *-------------------------------------------------------------------------
 */
@@ -134,8 +134,8 @@ DefineVirtualRelation(const RangeVar *relation, List *tlist, bool replace)
 	else
 	{
 		/*
-		 * now create the parameters for keys/inheritance etc. All of them
-		 * are nil...
+		 * now set the parameters for keys/inheritance etc. All of these
+		 * are uninteresting for views...
 		 */
 		createStmt->relation = (RangeVar *) relation;
 		createStmt->tableElts = attrList;
@@ -143,6 +143,7 @@ DefineVirtualRelation(const RangeVar *relation, List *tlist, bool replace)
 		createStmt->constraints = NIL;
 		createStmt->hasoids = MUST_NOT_HAVE_OIDS;
 		createStmt->oncommit = ONCOMMIT_NOOP;
+		createStmt->tablespacename = NULL;

 		/*
 		 * finally create the relation (this will error out if there's an