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.
author: Tom Lane <tgl@sss.pgh.pa.us> 2004-06-18 06:14:31 +0000
committer: Tom Lane <tgl@sss.pgh.pa.us> 2004-06-18 06:14:31 +0000
commit: 2467394ee1566e82d0314d12a0d1c0a5670a28c9 (patch)
tree: 57b87b8c181a9c3eb0f33bf775a5f31b9de8b890 /src
parent: 474875f4438ea0d18f9f4170117bc407e6812515 (diff)
download: postgresql-2467394ee1566e82d0314d12a0d1c0a5670a28c9.tar.gz
postgresql-2467394ee1566e82d0314d12a0d1c0a5670a28c9.zip
80 files changed, 2512 insertions, 722 deletions
diff --git a/src/backend/access/heap/heapam.c b/src/backend/access/heap/heapam.c
index cb1e6893053..8e595c18cb8 100644
--- a/src/backend/access/heap/heapam.c
+++ b/src/backend/access/heap/heapam.c
@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/access/heap/heapam.c,v 1.168 2004/05/27 17:12:37 tgl Exp $
+ *	  $PostgreSQL: pgsql/src/backend/access/heap/heapam.c,v 1.169 2004/06/18 06:13:09 tgl Exp $
  *
  *
  * INTERFACE ROUTINES
@@ -2630,8 +2630,8 @@ heap_undo(XLogRecPtr lsn, XLogRecord *record)
 static void
 out_target(char *buf, xl_heaptid *target)
 {
-	sprintf(buf + strlen(buf), "node %u/%u; tid %u/%u",
-			target->node.tblNode, target->node.relNode,
+	sprintf(buf + strlen(buf), "rel %u/%u/%u; tid %u/%u",
+			target->node.spcNode, target->node.dbNode, target->node.relNode,
 			ItemPointerGetBlockNumber(&(target->tid)),
 			ItemPointerGetOffsetNumber(&(target->tid)));
 }
@@ -2673,8 +2673,9 @@ heap_desc(char *buf, uint8 xl_info, char *rec)
 	{
 		xl_heap_clean *xlrec = (xl_heap_clean *) rec;
 
-		sprintf(buf + strlen(buf), "clean: node %u/%u; blk %u",
-				xlrec->node.tblNode, xlrec->node.relNode, xlrec->block);
+		sprintf(buf + strlen(buf), "clean: rel %u/%u/%u; blk %u",
+				xlrec->node.spcNode, xlrec->node.dbNode,
+				xlrec->node.relNode, xlrec->block);
 	}
 	else
 		strcat(buf, "UNKNOWN");
diff --git a/src/backend/access/nbtree/nbtxlog.c b/src/backend/access/nbtree/nbtxlog.c
index ed398b32da5..3cb2f3836d2 100644
--- a/src/backend/access/nbtree/nbtxlog.c
+++ b/src/backend/access/nbtree/nbtxlog.c
@@ -8,7 +8,7 @@
  * Portions Copyright (c) 1994, Regents of the University of California
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/access/nbtree/nbtxlog.c,v 1.13 2004/06/02 17:28:17 tgl Exp $
+ *	  $PostgreSQL: pgsql/src/backend/access/nbtree/nbtxlog.c,v 1.14 2004/06/18 06:13:11 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -803,8 +803,8 @@ btree_undo(XLogRecPtr lsn, XLogRecord *record)
 static void
 out_target(char *buf, xl_btreetid *target)
 {
-	sprintf(buf + strlen(buf), "node %u/%u; tid %u/%u",
-			target->node.tblNode, target->node.relNode,
+	sprintf(buf + strlen(buf), "rel %u/%u/%u; tid %u/%u",
+			target->node.spcNode, target->node.dbNode, target->node.relNode,
 			ItemPointerGetBlockNumber(&(target->tid)),
 			ItemPointerGetOffsetNumber(&(target->tid)));
 }
@@ -884,8 +884,9 @@ btree_desc(char *buf, uint8 xl_info, char *rec)
 			{
 				xl_btree_delete *xlrec = (xl_btree_delete *) rec;
 
-				sprintf(buf + strlen(buf), "delete: node %u/%u; blk %u",
-				 xlrec->node.tblNode, xlrec->node.relNode, xlrec->block);
+				sprintf(buf + strlen(buf), "delete: rel %u/%u/%u; blk %u",
+						xlrec->node.spcNode, xlrec->node.dbNode,
+						xlrec->node.relNode, xlrec->block);
 				break;
 			}
 		case XLOG_BTREE_DELETE_PAGE:
@@ -903,8 +904,9 @@ btree_desc(char *buf, uint8 xl_info, char *rec)
 			{
 				xl_btree_newroot *xlrec = (xl_btree_newroot *) rec;
 
-				sprintf(buf + strlen(buf), "newroot: node %u/%u; root %u lev %u",
-						xlrec->node.tblNode, xlrec->node.relNode,
+				sprintf(buf + strlen(buf), "newroot: rel %u/%u/%u; root %u lev %u",
+						xlrec->node.spcNode, xlrec->node.dbNode,
+						xlrec->node.relNode,
 						xlrec->rootblk, xlrec->level);
 				break;
 			}
@@ -912,8 +914,9 @@ btree_desc(char *buf, uint8 xl_info, char *rec)
 			{
 				xl_btree_newmeta *xlrec = (xl_btree_newmeta *) rec;
 
-				sprintf(buf + strlen(buf), "newmeta: node %u/%u; root %u lev %u fast %u lev %u",
-						xlrec->node.tblNode, xlrec->node.relNode,
+				sprintf(buf + strlen(buf), "newmeta: rel %u/%u/%u; root %u lev %u fast %u lev %u",
+						xlrec->node.spcNode, xlrec->node.dbNode,
+						xlrec->node.relNode,
 						xlrec->meta.root, xlrec->meta.level,
 						xlrec->meta.fastroot, xlrec->meta.fastlevel);
 				break;
@@ -922,9 +925,9 @@ btree_desc(char *buf, uint8 xl_info, char *rec)
 			{
 				xl_btree_newpage *xlrec = (xl_btree_newpage *) rec;
 
-				sprintf(buf + strlen(buf), "newpage: node %u/%u; page %u",
-						xlrec->node.tblNode, xlrec->node.relNode,
-						xlrec->blkno);
+				sprintf(buf + strlen(buf), "newpage: rel %u/%u/%u; page %u",
+						xlrec->node.spcNode, xlrec->node.dbNode,
+						xlrec->node.relNode, xlrec->blkno);
 				break;
 			}
 		default:
diff --git a/src/backend/access/transam/xlogutils.c b/src/backend/access/transam/xlogutils.c
index a7c8d3bf52b..a58902e76f6 100644
--- a/src/backend/access/transam/xlogutils.c
+++ b/src/backend/access/transam/xlogutils.c
@@ -6,7 +6,7 @@
  * Portions Copyright (c) 1996-2003, PostgreSQL Global Development Group
  * Portions Copyright (c) 1994, Regents of the University of California
  *
- * $PostgreSQL: pgsql/src/backend/access/transam/xlogutils.c,v 1.30 2004/02/11 22:55:24 tgl Exp $
+ * $PostgreSQL: pgsql/src/backend/access/transam/xlogutils.c,v 1.31 2004/06/18 06:13:15 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -354,7 +354,7 @@ XLogOpenRelation(bool redo, RmgrId rmid, RelFileNode rnode)
 		 * though, since we are presumably running by ourselves and can't
 		 * have any lock conflicts ...
 		 */
-		res->reldata.rd_lockInfo.lockRelId.dbId = rnode.tblNode;
+		res->reldata.rd_lockInfo.lockRelId.dbId = rnode.dbNode;
 		res->reldata.rd_lockInfo.lockRelId.relId = rnode.relNode;
 
 		hentry = (XLogRelCacheEntry *)
diff --git a/src/backend/bootstrap/bootparse.y b/src/backend/bootstrap/bootparse.y
index cbc8ff6e539..7b712bd555b 100644
--- a/src/backend/bootstrap/bootparse.y
+++ b/src/backend/bootstrap/bootparse.y
@@ -9,7 +9,7 @@
  *
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/bootstrap/bootparse.y,v 1.69 2004/06/03 02:08:02 tgl Exp $
+ *	  $PostgreSQL: pgsql/src/backend/bootstrap/bootparse.y,v 1.70 2004/06/18 06:13:17 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -31,6 +31,7 @@
 #include "catalog/pg_attribute.h"
 #include "catalog/pg_class.h"
 #include "catalog/pg_namespace.h"
+#include "catalog/pg_tablespace.h"
 #include "commands/defrem.h"
 #include "miscadmin.h"
 #include "nodes/makefuncs.h"
@@ -181,6 +182,7 @@ Boot_CreateStmt:
 
 						boot_reldesc = heap_create(LexIDStr($5),
 												   PG_CATALOG_NAMESPACE,
+												   $3 ? GLOBALTABLESPACE_OID : 0,
 												   tupdesc,
 												   $3,
 												   true,
@@ -193,6 +195,7 @@ Boot_CreateStmt:
 
 						id = heap_create_with_catalog(LexIDStr($5),
 													  PG_CATALOG_NAMESPACE,
+													  $3 ? GLOBALTABLESPACE_OID : 0,
 													  tupdesc,
 													  RELKIND_RELATION,
 													  $3,
@@ -239,6 +242,7 @@ Boot_DeclareIndexStmt:
 					DefineIndex(makeRangeVar(NULL, LexIDStr($5)),
 								LexIDStr($3),
 								LexIDStr($7),
+								NULL,
 								$9,
 								NULL, NIL,
 								false, false, false,
@@ -255,6 +259,7 @@ Boot_DeclareUniqueIndexStmt:
 					DefineIndex(makeRangeVar(NULL, LexIDStr($6)),
 								LexIDStr($4),
 								LexIDStr($8),
+								NULL,
 								$10,
 								NULL, NIL,
 								true, false, false,
diff --git a/src/backend/catalog/Makefile b/src/backend/catalog/Makefile
index 0a68cb661af..c63168a9bd5 100644
--- a/src/backend/catalog/Makefile
+++ b/src/backend/catalog/Makefile
@@ -2,7 +2,7 @@
 #
 # Makefile for backend/catalog
 #
-# $PostgreSQL: pgsql/src/backend/catalog/Makefile,v 1.50 2004/01/04 05:57:21 tgl Exp $
+# $PostgreSQL: pgsql/src/backend/catalog/Makefile,v 1.51 2004/06/18 06:13:19 tgl Exp $
 #
 #-------------------------------------------------------------------------
 
@@ -32,7 +32,7 @@ POSTGRES_BKI_SRCS := $(addprefix $(top_srcdir)/src/include/catalog/,\
 	pg_language.h pg_largeobject.h pg_aggregate.h pg_statistic.h \
 	pg_rewrite.h pg_trigger.h pg_listener.h pg_description.h pg_cast.h \
 	pg_namespace.h pg_conversion.h pg_database.h pg_shadow.h pg_group.h \
-	pg_depend.h indexing.h \
+	pg_tablespace.h pg_depend.h indexing.h \
     )
 
 pg_includes := $(sort -I$(top_srcdir)/src/include -I$(top_builddir)/src/include)
@@ -59,5 +59,5 @@ installdirs:
 uninstall-data:
 	rm -f $(addprefix $(DESTDIR)$(datadir)/, $(BKIFILES) system_views.sql information_schema.sql sql_features.txt)
 
-clean: 
+clean:
 	rm -f SUBSYS.o $(OBJS) $(BKIFILES)
diff --git a/src/backend/catalog/aclchk.c b/src/backend/catalog/aclchk.c
index de74a422b78..6c966b89b27 100644
--- a/src/backend/catalog/aclchk.c
+++ b/src/backend/catalog/aclchk.c
@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/catalog/aclchk.c,v 1.103 2004/06/01 21:49:22 tgl Exp $
+ *	  $PostgreSQL: pgsql/src/backend/catalog/aclchk.c,v 1.104 2004/06/18 06:13:19 tgl Exp $
  *
  * NOTES
  *	  See acl.h.
@@ -31,6 +31,7 @@
 #include "catalog/pg_operator.h"
 #include "catalog/pg_proc.h"
 #include "catalog/pg_shadow.h"
+#include "catalog/pg_tablespace.h"
 #include "catalog/pg_type.h"
 #include "miscadmin.h"
 #include "parser/parse_func.h"
@@ -45,6 +46,7 @@ static void ExecuteGrantStmt_Database(GrantStmt *stmt);
 static void ExecuteGrantStmt_Function(GrantStmt *stmt);
 static void ExecuteGrantStmt_Language(GrantStmt *stmt);
 static void ExecuteGrantStmt_Namespace(GrantStmt *stmt);
+static void ExecuteGrantStmt_Tablespace(GrantStmt *stmt);
 
 static const char *privilege_to_string(AclMode privilege);
 
@@ -207,12 +209,16 @@ ExecuteGrantStmt(GrantStmt *stmt)
 		case ACL_OBJECT_NAMESPACE:
 			ExecuteGrantStmt_Namespace(stmt);
 			break;
+		case ACL_OBJECT_TABLESPACE:
+			ExecuteGrantStmt_Tablespace(stmt);
+			break;
 		default:
 			elog(ERROR, "unrecognized GrantStmt.objtype: %d",
 				 (int) stmt->objtype);
 	}
 }
 
+
 static void
 ExecuteGrantStmt_Relation(GrantStmt *stmt)
 {
@@ -1009,6 +1015,163 @@ ExecuteGrantStmt_Namespace(GrantStmt *stmt)
 	}
 }
 
+static void
+ExecuteGrantStmt_Tablespace(GrantStmt *stmt)
+{
+	AclMode		privileges;
+	bool		all_privs;
+	ListCell   *i;
+
+	if (linitial_int(stmt->privileges) == ACL_ALL_RIGHTS)
+	{
+		all_privs = true;
+		privileges = ACL_ALL_RIGHTS_TABLESPACE;
+	}
+	else
+	{
+		all_privs = false;
+		privileges = ACL_NO_RIGHTS;
+		foreach(i, stmt->privileges)
+		{
+			AclMode		priv = lfirst_int(i);
+
+			if (priv & ~((AclMode) ACL_ALL_RIGHTS_TABLESPACE))
+				ereport(ERROR,
+						(errcode(ERRCODE_INVALID_GRANT_OPERATION),
+						 errmsg("invalid privilege type %s for tablespace",
+								privilege_to_string(priv))));
+			privileges |= priv;
+		}
+	}
+
+	foreach(i, stmt->objects)
+	{
+		char	   *spcname = strVal(lfirst(i));
+		Relation	relation;
+		ScanKeyData entry[1];
+		HeapScanDesc scan;
+		HeapTuple	tuple;
+		Form_pg_tablespace pg_tablespace_tuple;
+		Datum		aclDatum;
+		bool		isNull;
+		AclMode		my_goptions;
+		AclMode		this_privileges;
+		Acl		   *old_acl;
+		Acl		   *new_acl;
+		AclId		grantorId;
+		AclId		ownerId;
+		HeapTuple	newtuple;
+		Datum		values[Natts_pg_tablespace];
+		char		nulls[Natts_pg_tablespace];
+		char		replaces[Natts_pg_tablespace];
+
+		relation = heap_openr(TableSpaceRelationName, RowExclusiveLock);
+		ScanKeyInit(&entry[0],
+					Anum_pg_tablespace_spcname,
+					BTEqualStrategyNumber, F_NAMEEQ,
+					CStringGetDatum(spcname));
+		scan = heap_beginscan(relation, SnapshotNow, 1, entry);
+		tuple = heap_getnext(scan, ForwardScanDirection);
+		if (!HeapTupleIsValid(tuple))
+			ereport(ERROR,
+					(errcode(ERRCODE_UNDEFINED_OBJECT),
+					 errmsg("tablespace \"%s\" does not exist", spcname)));
+		pg_tablespace_tuple = (Form_pg_tablespace) GETSTRUCT(tuple);
+
+		ownerId = pg_tablespace_tuple->spcowner;
+		grantorId = select_grantor(ownerId);
+
+		/*
+		 * Must be owner or have some privilege on the object (per spec,
+		 * any privilege will get you by here).  The owner is always
+		 * treated as having all grant options.
+		 */
+		if (pg_tablespace_ownercheck(HeapTupleGetOid(tuple), GetUserId()))
+			my_goptions = ACL_ALL_RIGHTS_TABLESPACE;
+		else
+		{
+			AclMode		my_rights;
+
+			my_rights = pg_tablespace_aclmask(HeapTupleGetOid(tuple),
+											  GetUserId(),
+											  ACL_ALL_RIGHTS_TABLESPACE | ACL_GRANT_OPTION_FOR(ACL_ALL_RIGHTS_TABLESPACE),
+											  ACLMASK_ALL);
+			if (my_rights == ACL_NO_RIGHTS)
+				aclcheck_error(ACLCHECK_NO_PRIV, ACL_KIND_TABLESPACE,
+							   spcname);
+			my_goptions = ACL_OPTION_TO_PRIVS(my_rights);
+		}
+
+		/*
+		 * Restrict the operation to what we can actually grant or revoke,
+		 * and issue a warning if appropriate.  (For REVOKE this isn't quite
+		 * what the spec says to do: the spec seems to want a warning only
+		 * if no privilege bits actually change in the ACL.  In practice
+		 * that behavior seems much too noisy, as well as inconsistent with
+		 * the GRANT case.)
+		 */
+		this_privileges = privileges & my_goptions;
+		if (stmt->is_grant)
+		{
+			if (this_privileges == 0)
+				ereport(WARNING,
+						(errcode(ERRCODE_WARNING_PRIVILEGE_NOT_GRANTED),
+						 errmsg("no privileges were granted")));
+			else if (!all_privs && this_privileges != privileges)
+				ereport(WARNING,
+						(errcode(ERRCODE_WARNING_PRIVILEGE_NOT_GRANTED),
+						 errmsg("not all privileges were granted")));
+		}
+		else
+		{
+			if (this_privileges == 0)
+				ereport(WARNING,
+						(errcode(ERRCODE_WARNING_PRIVILEGE_NOT_REVOKED),
+						 errmsg("no privileges could be revoked")));
+			else if (!all_privs && this_privileges != privileges)
+				ereport(WARNING,
+						(errcode(ERRCODE_WARNING_PRIVILEGE_NOT_REVOKED),
+						 errmsg("not all privileges could be revoked")));
+		}
+
+		/*
+		 * If there's no ACL, substitute the proper default.
+		 */
+		aclDatum = heap_getattr(tuple, Anum_pg_tablespace_spcacl,
+								RelationGetDescr(relation), &isNull);
+		if (isNull)
+			old_acl = acldefault(ACL_OBJECT_TABLESPACE, ownerId);
+		else
+			/* get a detoasted copy of the ACL */
+			old_acl = DatumGetAclPCopy(aclDatum);
+
+		new_acl = merge_acl_with_grant(old_acl, stmt->is_grant,
+									   stmt->grant_option, stmt->behavior,
+									   stmt->grantees, this_privileges,
+									   grantorId, ownerId);
+
+		/* finished building new ACL value, now insert it */
+		MemSet(values, 0, sizeof(values));
+		MemSet(nulls, ' ', sizeof(nulls));
+		MemSet(replaces, ' ', sizeof(replaces));
+
+		replaces[Anum_pg_tablespace_spcacl - 1] = 'r';
+		values[Anum_pg_tablespace_spcacl - 1] = PointerGetDatum(new_acl);
+
+		newtuple = heap_modifytuple(tuple, relation, values, nulls, replaces);
+
+		simple_heap_update(relation, &newtuple->t_self, newtuple);
+
+		/* keep the catalog indexes up to date */
+		CatalogUpdateIndexes(relation, newtuple);
+
+		pfree(new_acl);
+
+		heap_endscan(scan);
+		heap_close(relation, RowExclusiveLock);
+	}
+}
+
 
 static const char *
 privilege_to_string(AclMode privilege)
@@ -1112,7 +1275,9 @@ static const char *const no_priv_msg[MAX_ACL_KIND] =
 	/* ACL_KIND_OPCLASS */
 	gettext_noop("permission denied for operator class %s"),
 	/* ACL_KIND_CONVERSION */
-	gettext_noop("permission denied for conversion %s")
+	gettext_noop("permission denied for conversion %s"),
+	/* ACL_KIND_TABLESPACE */
+	gettext_noop("permission denied for tablespace %s")
 };
 
 static const char *const not_owner_msg[MAX_ACL_KIND] =
@@ -1134,7 +1299,9 @@ static const char *const not_owner_msg[MAX_ACL_KIND] =
 	/* ACL_KIND_OPCLASS */
 	gettext_noop("must be owner of operator class %s"),
 	/* ACL_KIND_CONVERSION */
-	gettext_noop("must be owner of conversion %s")
+	gettext_noop("must be owner of conversion %s"),
+	/* ACL_KIND_TABLESPACE */
+	gettext_noop("must be owner of tablespace %s")
 };
 
 
@@ -1545,6 +1712,80 @@ pg_namespace_aclmask(Oid nsp_oid, AclId userid,
 	return result;
 }
 
+/*
+ * Exported routine for examining a user's privileges for a tablespace
+ */
+AclMode
+pg_tablespace_aclmask(Oid spc_oid, AclId userid,
+					  AclMode mask, AclMaskHow how)
+{
+	AclMode		result;
+	Relation	pg_tablespace;
+	ScanKeyData entry[1];
+	HeapScanDesc scan;
+	HeapTuple	tuple;
+	Datum		aclDatum;
+	bool		isNull;
+	Acl		   *acl;
+	AclId		ownerId;
+
+	/*
+	 * Only shared relations can be stored in global space; don't let
+	 * even superusers override this
+	 */
+	if (spc_oid == GLOBALTABLESPACE_OID && !IsBootstrapProcessingMode())
+		return 0;
+
+	/* Otherwise, superusers bypass all permission checking. */
+	if (superuser_arg(userid))
+		return mask;
+
+	/*
+	 * Get the tablespace's ACL from pg_tablespace
+	 *
+	 * There's no syscache for pg_tablespace, so must look the hard way
+	 */
+	pg_tablespace = heap_openr(TableSpaceRelationName, AccessShareLock);
+	ScanKeyInit(&entry[0],
+				ObjectIdAttributeNumber,
+				BTEqualStrategyNumber, F_OIDEQ,
+				ObjectIdGetDatum(spc_oid));
+	scan = heap_beginscan(pg_tablespace, SnapshotNow, 1, entry);
+	tuple = heap_getnext(scan, ForwardScanDirection);
+	if (!HeapTupleIsValid(tuple))
+		ereport(ERROR,
+				(errcode(ERRCODE_UNDEFINED_OBJECT),
+				 errmsg("tablespace with OID %u does not exist", spc_oid)));
+
+	ownerId = ((Form_pg_tablespace) GETSTRUCT(tuple))->spcowner;
+
+	aclDatum = heap_getattr(tuple, Anum_pg_tablespace_spcacl,
+							RelationGetDescr(pg_tablespace), &isNull);
+
+	if (isNull)
+	{
+		/* No ACL, so build default ACL */
+		acl = acldefault(ACL_OBJECT_TABLESPACE, ownerId);
+		aclDatum = (Datum) 0;
+	}
+	else
+	{
+		/* detoast ACL if necessary */
+		acl = DatumGetAclP(aclDatum);
+	}
+
+	result = aclmask(acl, userid, ownerId, mask, how);
+
+	/* if we have a detoasted copy, free it */
+	if (acl && (Pointer) acl != DatumGetPointer(aclDatum))
+		pfree(acl);
+
+	heap_endscan(scan);
+	heap_close(pg_tablespace, AccessShareLock);
+
+	return result;
+}
+
 
 /*
  * Exported routine for checking a user's access privileges to a table
@@ -1610,6 +1851,18 @@ pg_namespace_aclcheck(Oid nsp_oid, AclId userid, AclMode mode)
 		return ACLCHECK_NO_PRIV;
 }
 
+/*
+ * Exported routine for checking a user's access privileges to a tablespace
+ */
+AclResult
+pg_tablespace_aclcheck(Oid spc_oid, AclId userid, AclMode mode)
+{
+	if (pg_tablespace_aclmask(spc_oid, userid, mode, ACLMASK_ANY) != 0)
+		return ACLCHECK_OK;
+	else
+		return ACLCHECK_NO_PRIV;
+}
+
 
 /*
  * Ownership check for a relation (specified by OID).
@@ -1752,6 +2005,45 @@ pg_namespace_ownercheck(Oid nsp_oid, AclId userid)
 }
 
 /*
+ * Ownership check for a tablespace (specified by OID).
+ */
+bool
+pg_tablespace_ownercheck(Oid spc_oid, AclId userid)
+{
+	Relation	pg_tablespace;
+	ScanKeyData entry[1];
+	HeapScanDesc scan;
+	HeapTuple	spctuple;
+	int32		spcowner;
+
+	/* Superusers bypass all permission checking. */
+	if (superuser_arg(userid))
+		return true;
+
+	/* There's no syscache for pg_tablespace, so must look the hard way */
+	pg_tablespace = heap_openr(TableSpaceRelationName, AccessShareLock);
+	ScanKeyInit(&entry[0],
+				ObjectIdAttributeNumber,
+				BTEqualStrategyNumber, F_OIDEQ,
+				ObjectIdGetDatum(spc_oid));
+	scan = heap_beginscan(pg_tablespace, SnapshotNow, 1, entry);
+
+	spctuple = heap_getnext(scan, ForwardScanDirection);
+
+	if (!HeapTupleIsValid(spctuple))
+		ereport(ERROR,
+				(errcode(ERRCODE_UNDEFINED_OBJECT),
+				 errmsg("tablespace with OID %u does not exist", spc_oid)));
+
+	spcowner = ((Form_pg_tablespace) GETSTRUCT(spctuple))->spcowner;
+
+	heap_endscan(scan);
+	heap_close(pg_tablespace, AccessShareLock);
+
+	return userid == spcowner;
+}
+
+/*
  * Ownership check for an operator class (specified by OID).
  */
 bool
@@ -1780,9 +2072,8 @@ pg_opclass_ownercheck(Oid opc_oid, AclId userid)
 	return userid == owner_id;
 }
 
-
 /*
- * Ownership check for database (specified as OID)
+ * Ownership check for a database (specified by OID).
  */
 bool
 pg_database_ownercheck(Oid db_oid, AclId userid)
diff --git a/src/backend/catalog/catalog.c b/src/backend/catalog/catalog.c
index afea81a976c..32d4d33d89a 100644
--- a/src/backend/catalog/catalog.c
+++ b/src/backend/catalog/catalog.c
@@ -9,7 +9,7 @@
  *
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/catalog/catalog.c,v 1.51 2004/01/06 18:07:31 neilc Exp $
+ *	  $PostgreSQL: pgsql/src/backend/catalog/catalog.c,v 1.52 2004/06/18 06:13:19 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -20,30 +20,48 @@
 #include "catalog/catalog.h"
 #include "catalog/catname.h"
 #include "catalog/pg_namespace.h"
+#include "catalog/pg_tablespace.h"
 #include "miscadmin.h"
 
 
+#define OIDCHARS	10			/* max chars printed by %u */
+
+
 /*
  * relpath			- construct path to a relation's file
  *
  * Result is a palloc'd string.
  */
-
 char *
 relpath(RelFileNode rnode)
 {
+	int			pathlen;
 	char	   *path;
 
-	if (rnode.tblNode == (Oid) 0)		/* "global tablespace" */
+	if (rnode.spcNode == GLOBALTABLESPACE_OID)
 	{
 		/* Shared system relations live in {datadir}/global */
-		path = (char *) palloc(strlen(DataDir) + 8 + sizeof(NameData) + 1);
-		sprintf(path, "%s/global/%u", DataDir, rnode.relNode);
+		Assert(rnode.dbNode == 0);
+		pathlen = strlen(DataDir) + 8 + OIDCHARS + 1;
+		path = (char *) palloc(pathlen);
+		snprintf(path, pathlen, "%s/global/%u",
+				 DataDir, rnode.relNode);
+	}
+	else if (rnode.spcNode == DEFAULTTABLESPACE_OID)
+	{
+		/* The default tablespace is {datadir}/base */
+		pathlen = strlen(DataDir) + 6 + OIDCHARS + 1 + OIDCHARS + 1;
+		path = (char *) palloc(pathlen);
+		snprintf(path, pathlen, "%s/base/%u/%u",
+				 DataDir, rnode.dbNode, rnode.relNode);
 	}
 	else
 	{
-		path = (char *) palloc(strlen(DataDir) + 6 + 2 * sizeof(NameData) + 3);
-		sprintf(path, "%s/base/%u/%u", DataDir, rnode.tblNode, rnode.relNode);
+		/* All other tablespaces are accessed via symlinks */
+		pathlen = strlen(DataDir) + 16 + OIDCHARS + 1 + OIDCHARS + 1 + OIDCHARS + 1;
+		path = (char *) palloc(pathlen);
+		snprintf(path, pathlen, "%s/pg_tablespaces/%u/%u/%u",
+				 DataDir, rnode.spcNode, rnode.dbNode, rnode.relNode);
 	}
 	return path;
 }
@@ -52,23 +70,39 @@ relpath(RelFileNode rnode)
  * GetDatabasePath			- construct path to a database dir
  *
  * Result is a palloc'd string.
+ *
+ * XXX this must agree with relpath()!
  */
-
 char *
-GetDatabasePath(Oid tblNode)
+GetDatabasePath(Oid dbNode, Oid spcNode)
 {
+	int			pathlen;
 	char	   *path;
 
-	if (tblNode == (Oid) 0)		/* "global tablespace" */
+	if (spcNode == GLOBALTABLESPACE_OID)
 	{
 		/* Shared system relations live in {datadir}/global */
-		path = (char *) palloc(strlen(DataDir) + 8);
-		sprintf(path, "%s/global", DataDir);
+		Assert(dbNode == 0);
+		pathlen = strlen(DataDir) + 7 + 1;
+		path = (char *) palloc(pathlen);
+		snprintf(path, pathlen, "%s/global",
+				 DataDir);
+	}
+	else if (spcNode == DEFAULTTABLESPACE_OID)
+	{
+		/* The default tablespace is {datadir}/base */
+		pathlen = strlen(DataDir) + 6 + OIDCHARS + 1;
+		path = (char *) palloc(pathlen);
+		snprintf(path, pathlen, "%s/base/%u",
+				 DataDir, dbNode);
 	}
 	else
 	{
-		path = (char *) palloc(strlen(DataDir) + 6 + sizeof(NameData) + 1);
-		sprintf(path, "%s/base/%u", DataDir, tblNode);
+		/* All other tablespaces are accessed via symlinks */
+		pathlen = strlen(DataDir) + 16 + OIDCHARS + 1 + OIDCHARS + 1;
+		path = (char *) palloc(pathlen);
+		snprintf(path, pathlen, "%s/pg_tablespaces/%u/%u",
+				 DataDir, spcNode, dbNode);
 	}
 	return path;
 }
diff --git a/src/backend/catalog/heap.c b/src/backend/catalog/heap.c
index a4cc33d4c9d..68ea6f45f69 100644
--- a/src/backend/catalog/heap.c
+++ b/src/backend/catalog/heap.c
@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/catalog/heap.c,v 1.270 2004/06/10 17:55:53 tgl Exp $
+ *	  $PostgreSQL: pgsql/src/backend/catalog/heap.c,v 1.271 2004/06/18 06:13:19 tgl Exp $
  *
  *
  * INTERFACE ROUTINES
@@ -43,6 +43,7 @@
 #include "catalog/pg_statistic.h"
 #include "catalog/pg_type.h"
 #include "commands/tablecmds.h"
+#include "commands/tablespace.h"
 #include "commands/trigger.h"
 #include "miscadmin.h"
 #include "nodes/makefuncs.h"
@@ -203,15 +204,14 @@ SystemAttributeByName(const char *attname, bool relhasoids)
 Relation
 heap_create(const char *relname,
 			Oid relnamespace,
+			Oid reltablespace,
 			TupleDesc tupDesc,
 			bool shared_relation,
 			bool storage_create,
 			bool allow_system_table_mods)
 {
 	Oid			relid;
-	Oid			dbid = shared_relation ? InvalidOid : MyDatabaseId;
 	bool		nailme = false;
-	RelFileNode rnode;
 	Relation	rel;
 
 	/*
@@ -260,6 +260,8 @@ heap_create(const char *relname,
 			relid = RelOid_pg_group;
 		else if (strcmp(DatabaseRelationName, relname) == 0)
 			relid = RelOid_pg_database;
+		else if (strcmp(TableSpaceRelationName, relname) == 0)
+			relid = RelOid_pg_tablespace;
 		else
 			relid = newoid();
 	}
@@ -267,20 +269,14 @@ heap_create(const char *relname,
 		relid = newoid();
 
 	/*
-	 * For now, the physical identifier of the relation is the same as the
-	 * logical identifier.
-	 */
-	rnode.tblNode = dbid;
-	rnode.relNode = relid;
-
-	/*
 	 * build the relcache entry.
 	 */
 	rel = RelationBuildLocalRelation(relname,
 									 relnamespace,
 									 tupDesc,
-									 relid, dbid,
-									 rnode,
+									 relid,
+									 reltablespace,
+									 shared_relation,
 									 nailme);
 
 	/*
@@ -296,6 +292,16 @@ heap_create(const char *relname,
 void
 heap_storage_create(Relation rel)
 {
+	/*
+	 * We may be using the target table space for the first time in this
+	 * database, so create a per-database subdirectory if needed.
+	 *
+	 * XXX it might be better to do this right in smgrcreate...
+	 */
+	TablespaceCreateDbspace(rel->rd_node.spcNode, rel->rd_node.dbNode);
+	/*
+	 * Now we can make the file.
+	 */
 	Assert(rel->rd_smgr == NULL);
 	rel->rd_smgr = smgropen(rel->rd_node);
 	smgrcreate(rel->rd_smgr, rel->rd_istemp, false);
@@ -692,6 +698,7 @@ AddNewRelationType(const char *typeName,
 Oid
 heap_create_with_catalog(const char *relname,
 						 Oid relnamespace,
+						 Oid reltablespace,
 						 TupleDesc tupdesc,
 						 char relkind,
 						 bool shared_relation,
@@ -726,6 +733,7 @@ heap_create_with_catalog(const char *relname,
 	 */
 	new_rel_desc = heap_create(relname,
 							   relnamespace,
+							   reltablespace,
 							   tupdesc,
 							   shared_relation,
 							   (relkind != RELKIND_VIEW &&
diff --git a/src/backend/catalog/index.c b/src/backend/catalog/index.c
index 8ada0915bd2..581799fc5f5 100644
--- a/src/backend/catalog/index.c
+++ b/src/backend/catalog/index.c
@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/catalog/index.c,v 1.233 2004/05/31 19:24:05 tgl Exp $
+ *	  $PostgreSQL: pgsql/src/backend/catalog/index.c,v 1.234 2004/06/18 06:13:19 tgl Exp $
  *
  *
  * INTERFACE ROUTINES
@@ -467,6 +467,7 @@ index_create(Oid heapRelationId,
 			 const char *indexRelationName,
 			 IndexInfo *indexInfo,
 			 Oid accessMethodObjectId,
+			 Oid tableSpaceId,
 			 Oid *classObjectId,
 			 bool primary,
 			 bool isconstraint,
@@ -539,6 +540,7 @@ index_create(Oid heapRelationId,
 	 */
 	indexRelation = heap_create(indexRelationName,
 								namespaceId,
+								tableSpaceId,
 								indexTupDesc,
 								shared_relation,
 								true,
diff --git a/src/backend/catalog/namespace.c b/src/backend/catalog/namespace.c
index c7535508a73..b412023fe28 100644
--- a/src/backend/catalog/namespace.c
+++ b/src/backend/catalog/namespace.c
@@ -13,7 +13,7 @@
  * Portions Copyright (c) 1994, Regents of the University of California
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/catalog/namespace.c,v 1.66 2004/05/28 16:17:14 tgl Exp $
+ *	  $PostgreSQL: pgsql/src/backend/catalog/namespace.c,v 1.67 2004/06/18 06:13:19 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -1668,7 +1668,7 @@ InitTempTableNamespace(void)
 		 * that access the temp namespace for my own backend skip
 		 * permissions checks on it.
 		 */
-		namespaceId = NamespaceCreate(namespaceName, BOOTSTRAP_USESYSID);
+		namespaceId = NamespaceCreate(namespaceName, BOOTSTRAP_USESYSID, 0);
 		/* Advance command counter to make namespace visible */
 		CommandCounterIncrement();
 	}
diff --git a/src/backend/catalog/pg_namespace.c b/src/backend/catalog/pg_namespace.c
index c3546f9068f..c600ac2a84d 100644
--- a/src/backend/catalog/pg_namespace.c
+++ b/src/backend/catalog/pg_namespace.c
@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/catalog/pg_namespace.c,v 1.8 2003/11/29 19:51:46 pgsql Exp $
+ *	  $PostgreSQL: pgsql/src/backend/catalog/pg_namespace.c,v 1.9 2004/06/18 06:13:19 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -27,7 +27,7 @@
  * ---------------
  */
 Oid
-NamespaceCreate(const char *nspName, int32 ownerSysId)
+NamespaceCreate(const char *nspName, int32 ownerSysId, Oid nspTablespace)
 {
 	Relation	nspdesc;
 	HeapTuple	tup;
@@ -59,6 +59,7 @@ NamespaceCreate(const char *nspName, int32 ownerSysId)
 	namestrcpy(&nname, nspName);
 	values[Anum_pg_namespace_nspname - 1] = NameGetDatum(&nname);
 	values[Anum_pg_namespace_nspowner - 1] = Int32GetDatum(ownerSysId);
+	values[Anum_pg_namespace_nsptablespace - 1] = Int32GetDatum(nspTablespace);
 	nulls[Anum_pg_namespace_nspacl - 1] = 'n';
 
 	nspdesc = heap_openr(NamespaceRelationName, RowExclusiveLock);
diff --git a/src/backend/commands/Makefile b/src/backend/commands/Makefile
index 0c450beac3c..644fd1d655d 100644
--- 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
 
diff --git a/src/backend/commands/cluster.c b/src/backend/commands/cluster.c
index 0483eaf2d3c..d81771c7a04 100644
--- 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,
diff --git a/src/backend/commands/dbcommands.c b/src/backend/commands/dbcommands.c
index 337ec5395fb..8fbebecd874 100644
--- 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,36 +78,41 @@ 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 (downer)
+			if (dtablespacename)
 				ereport(ERROR,
 						(errcode(ERRCODE_SYNTAX_ERROR),
 						 errmsg("conflicting or redundant options")));
-			downer = defel;
+			dtablespacename = defel;
 		}
-		else if (strcmp(defel->defname, "location") == 0)
+		else if (strcmp(defel->defname, "owner") == 0)
 		{
-			if (dpath)
+			if (downer)
 				ereport(ERROR,
 						(errcode(ERRCODE_SYNTAX_ERROR),
 						 errmsg("conflicting or redundant options")));
-			dpath = defel;
+			downer = defel;
 		}
 		else if (strcmp(defel->defname, "template") == 0)
 		{
@@ -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)));
@@ -247,14 +240,6 @@ createdb(const CreatedbStmt *stmt)
 	}
 
 	/*
-	 * 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).
 	 * Otherwise we might copy inconsistent data.  This check is not
@@ -276,6 +261,33 @@ 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.
@@ -283,39 +295,6 @@ createdb(const CreatedbStmt *stmt)
 	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
 	 * the source database...)
@@ -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.
+	 * Iterate through all tablespaces of the template database, and
+	 * copy each one to the new database.
+	 *
+	 * 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.
 	 */
-	target_dir = alt_loc ? alt_loc : nominal_loc;
+	rel = heap_openr(TableSpaceRelationName, AccessShareLock);
+	scan = heap_beginscan(rel, SnapshotNow, 0, NULL);
+	while ((tuple = heap_getnext(scan, ForwardScanDirection)) != NULL)
+	{
+		Oid		srctablespace = HeapTupleGetOid(tuple);
+		Oid		dsttablespace;
+		char   *srcpath;
+		char   *dstpath;
+		struct stat st;
 
-	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)));
+		/* No need to copy global tablespace */
+		if (srctablespace == GLOBALTABLESPACE_OID)
+			continue;
 
-	/* Make the symlink, if needed */
-	if (alt_loc)
-	{
-#ifdef HAVE_SYMLINK				/* already throws error above */
-		if (symlink(alt_loc, nominal_loc) != 0)
-#endif
+		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,
-					(errcode_for_file_access(),
-					 errmsg("could not link file \"%s\" to \"%s\": %m",
-							nominal_loc, alt_loc)));
-	}
+					(errmsg("could not initialize database directory"),
+					 errdetail("Directory \"%s\" already exists.", dstpath)));
+		}
 
-	/*
-	 * Copy the template database 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.
-	 */
 #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)
-	{
-		if (remove_dbdirs(nominal_loc, alt_loc))
+		/*
+		 * 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;
-
-	if (dbpath == NULL || dbpath[0] == '\0')
-		return NULL;
-
-	if (first_dir_separator(dbpath))
-	{
-		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
+	Relation rel;
+	HeapScanDesc scan;
+	HeapTuple tuple;
+	char buf[MAXPGPATH + 100];
+
+	rel = heap_openr(TableSpaceRelationName, AccessShareLock);
+	scan = heap_beginscan(rel, SnapshotNow, 0, NULL);
+	while ((tuple = heap_getnext(scan, ForwardScanDirection)) != NULL)
 	{
-		/* must be environment variable */
-		char	   *var = getenv(dbpath);
-
-		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;
-	}
+		Oid		dsttablespace = HeapTupleGetOid(tuple);
+		char   *dstpath;
+		struct stat st;
 
-	len = strlen(prefix) + 6 + sizeof(Oid) * 8 + 1;
-	if (len >= MAXPGPATH - 100)
-		ereport(ERROR,
-				(errcode(ERRCODE_INVALID_NAME),
-				 errmsg("alternative path is too long")));
+		/* Don't mess with the global tablespace */
+		if (dsttablespace == GLOBALTABLESPACE_OID)
+			continue;
 
-	ret = palloc(len);
-	snprintf(ret, len, "%s/base/%u", prefix, dboid);
+		dstpath = GetDatabasePath(db_id, dsttablespace);
 
-	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.
  */
diff --git a/src/backend/commands/indexcmds.c b/src/backend/commands/indexcmds.c
index d141c83c1bf..43d9aabd046 100644
--- 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);
 
diff --git a/src/backend/commands/schemacmds.c b/src/backend/commands/schemacmds.c
index 18a212271ea..8366ea634a0 100644
--- 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();
diff --git a/src/backend/commands/sequence.c b/src/backend/commands/sequence.c
index 351813b5dec..1b6538b539c 100644
--- 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);
 }
diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c
index 3979335313b..8fd07e396ae 100644
--- 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;
@@ -302,6 +304,31 @@ DefineRelation(CreateStmt *stmt, char relkind)
 	}
 
 	/*
+	 * 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);
 
 	/*
diff --git a/src/backend/commands/tablespace.c b/src/backend/commands/tablespace.c
new file mode 100644
index 00000000000..d412389bb86
--- /dev/null
+++ 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;
+}
diff --git a/src/backend/commands/typecmds.c b/src/backend/commands/typecmds.c
index a64617d08d4..d087ad8895c 100644
--- 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...
diff --git a/src/backend/commands/view.c b/src/backend/commands/view.c
index 4c27bd67199..67971219699 100644
--- 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
diff --git a/src/backend/executor/execMain.c b/src/backend/executor/execMain.c
index 4099ef0a791..6f0f22a3221 100644
--- a/src/backend/executor/execMain.c
+++ b/src/backend/executor/execMain.c
@@ -26,7 +26,7 @@
  *
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/executor/execMain.c,v 1.233 2004/05/30 23:40:26 neilc Exp $
+ *	  $PostgreSQL: pgsql/src/backend/executor/execMain.c,v 1.234 2004/06/18 06:13:26 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -779,6 +779,7 @@ InitPlan(QueryDesc *queryDesc, bool explainOnly)
 
 		intoRelationId = heap_create_with_catalog(intoName,
 												  namespaceId,
+												  InvalidOid,
 												  tupdesc,
 												  RELKIND_RELATION,
 												  false,
diff --git a/src/backend/nodes/copyfuncs.c b/src/backend/nodes/copyfuncs.c
index bfcb82447d8..90983e6db0b 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
- *	  $PostgreSQL: pgsql/src/backend/nodes/copyfuncs.c,v 1.285 2004/06/09 19:08:15 tgl Exp $
+ *	  $PostgreSQL: pgsql/src/backend/nodes/copyfuncs.c,v 1.286 2004/06/18 06:13:28 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -1753,6 +1753,7 @@ _copyCreateStmt(CreateStmt *from)
 	COPY_NODE_FIELD(constraints);
 	COPY_SCALAR_FIELD(hasoids);
 	COPY_SCALAR_FIELD(oncommit);
+	COPY_STRING_FIELD(tablespacename);
 
 	return newnode;
 }
@@ -1836,6 +1837,7 @@ _copyIndexStmt(IndexStmt *from)
 	COPY_STRING_FIELD(idxname);
 	COPY_NODE_FIELD(relation);
 	COPY_STRING_FIELD(accessMethod);
+	COPY_STRING_FIELD(tableSpace);
 	COPY_NODE_FIELD(indexParams);
 	COPY_NODE_FIELD(whereClause);
 	COPY_NODE_FIELD(rangetable);
@@ -2146,6 +2148,7 @@ _copyCreateSeqStmt(CreateSeqStmt *from)
 
 	COPY_NODE_FIELD(sequence);
 	COPY_NODE_FIELD(options);
+	COPY_STRING_FIELD(tablespacename);
 
 	return newnode;
 }
@@ -2193,6 +2196,28 @@ _copyVariableResetStmt(VariableResetStmt *from)
 	return newnode;
 }
 
+static CreateTableSpaceStmt *
+_copyCreateTableSpaceStmt(CreateTableSpaceStmt *from)
+{
+	CreateTableSpaceStmt *newnode = makeNode(CreateTableSpaceStmt);
+
+	COPY_STRING_FIELD(tablespacename);
+	COPY_STRING_FIELD(owner);
+	COPY_STRING_FIELD(location);
+
+	return newnode;
+}
+
+static DropTableSpaceStmt *
+_copyDropTableSpaceStmt(DropTableSpaceStmt *from)
+{
+	DropTableSpaceStmt *newnode = makeNode(DropTableSpaceStmt);
+
+	COPY_STRING_FIELD(tablespacename);
+
+	return newnode;
+}
+
 static CreateTrigStmt *
 _copyCreateTrigStmt(CreateTrigStmt *from)
 {
@@ -2371,6 +2396,7 @@ _copyCreateSchemaStmt(CreateSchemaStmt *from)
 
 	COPY_STRING_FIELD(schemaname);
 	COPY_STRING_FIELD(authid);
+	COPY_STRING_FIELD(tablespacename);
 	COPY_NODE_FIELD(schemaElts);
 
 	return newnode;
@@ -2914,6 +2940,12 @@ copyObject(void *from)
 		case T_VariableResetStmt:
 			retval = _copyVariableResetStmt(from);
 			break;
+		case T_CreateTableSpaceStmt:
+			retval = _copyCreateTableSpaceStmt(from);
+			break;
+		case T_DropTableSpaceStmt:
+			retval = _copyDropTableSpaceStmt(from);
+			break;
 		case T_CreateTrigStmt:
 			retval = _copyCreateTrigStmt(from);
 			break;
diff --git a/src/backend/nodes/equalfuncs.c b/src/backend/nodes/equalfuncs.c
index fec550836b3..47ec3157727 100644
--- a/src/backend/nodes/equalfuncs.c
+++ b/src/backend/nodes/equalfuncs.c
@@ -18,7 +18,7 @@
  * Portions Copyright (c) 1994, Regents of the University of California
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/nodes/equalfuncs.c,v 1.224 2004/06/09 19:08:15 tgl Exp $
+ *	  $PostgreSQL: pgsql/src/backend/nodes/equalfuncs.c,v 1.225 2004/06/18 06:13:28 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -835,6 +835,7 @@ _equalCreateStmt(CreateStmt *a, CreateStmt *b)
 	COMPARE_NODE_FIELD(constraints);
 	COMPARE_SCALAR_FIELD(hasoids);
 	COMPARE_SCALAR_FIELD(oncommit);
+	COMPARE_STRING_FIELD(tablespacename);
 
 	return true;
 }
@@ -904,6 +905,7 @@ _equalIndexStmt(IndexStmt *a, IndexStmt *b)
 	COMPARE_STRING_FIELD(idxname);
 	COMPARE_NODE_FIELD(relation);
 	COMPARE_STRING_FIELD(accessMethod);
+	COMPARE_STRING_FIELD(tableSpace);
 	COMPARE_NODE_FIELD(indexParams);
 	COMPARE_NODE_FIELD(whereClause);
 	COMPARE_NODE_FIELD(rangetable);
@@ -1164,6 +1166,7 @@ _equalCreateSeqStmt(CreateSeqStmt *a, CreateSeqStmt *b)
 {
 	COMPARE_NODE_FIELD(sequence);
 	COMPARE_NODE_FIELD(options);
+	COMPARE_STRING_FIELD(tablespacename);
 
 	return true;
 }
@@ -1204,6 +1207,24 @@ _equalVariableResetStmt(VariableResetStmt *a, VariableResetStmt *b)
 }
 
 static bool
+_equalCreateTableSpaceStmt(CreateTableSpaceStmt *a, CreateTableSpaceStmt *b)
+{
+	COMPARE_STRING_FIELD(tablespacename);
+	COMPARE_STRING_FIELD(owner);
+	COMPARE_STRING_FIELD(location);
+
+	return true;
+}
+
+static bool
+_equalDropTableSpaceStmt(DropTableSpaceStmt *a, DropTableSpaceStmt *b)
+{
+	COMPARE_STRING_FIELD(tablespacename);
+
+	return true;
+}
+
+static bool
 _equalCreateTrigStmt(CreateTrigStmt *a, CreateTrigStmt *b)
 {
 	COMPARE_STRING_FIELD(trigname);
@@ -1352,6 +1373,7 @@ _equalCreateSchemaStmt(CreateSchemaStmt *a, CreateSchemaStmt *b)
 {
 	COMPARE_STRING_FIELD(schemaname);
 	COMPARE_STRING_FIELD(authid);
+	COMPARE_STRING_FIELD(tablespacename);
 	COMPARE_NODE_FIELD(schemaElts);
 
 	return true;
@@ -2052,6 +2074,12 @@ equal(void *a, void *b)
 		case T_VariableResetStmt:
 			retval = _equalVariableResetStmt(a, b);
 			break;
+		case T_CreateTableSpaceStmt:
+			retval = _equalCreateTableSpaceStmt(a, b);
+			break;
+		case T_DropTableSpaceStmt:
+			retval = _equalDropTableSpaceStmt(a, b);
+			break;
 		case T_CreateTrigStmt:
 			retval = _equalCreateTrigStmt(a, b);
 			break;
diff --git a/src/backend/nodes/outfuncs.c b/src/backend/nodes/outfuncs.c
index 1984ced756f..087ebb39a15 100644
--- a/src/backend/nodes/outfuncs.c
+++ b/src/backend/nodes/outfuncs.c
@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/nodes/outfuncs.c,v 1.239 2004/06/09 19:08:15 tgl Exp $
+ *	  $PostgreSQL: pgsql/src/backend/nodes/outfuncs.c,v 1.240 2004/06/18 06:13:28 tgl Exp $
  *
  * NOTES
  *	  Every node type that can appear in stored rules' parsetrees *must*
@@ -1128,7 +1128,7 @@ _outInClauseInfo(StringInfo str, InClauseInfo *node)
 static void
 _outCreateStmt(StringInfo str, CreateStmt *node)
 {
-	WRITE_NODE_TYPE("CREATE");
+	WRITE_NODE_TYPE("CREATESTMT");
 
 	WRITE_NODE_FIELD(relation);
 	WRITE_NODE_FIELD(tableElts);
@@ -1136,16 +1136,18 @@ _outCreateStmt(StringInfo str, CreateStmt *node)
 	WRITE_NODE_FIELD(constraints);
 	WRITE_ENUM_FIELD(hasoids, ContainsOids);
 	WRITE_ENUM_FIELD(oncommit, OnCommitAction);
+	WRITE_STRING_FIELD(tablespacename);
 }
 
 static void
 _outIndexStmt(StringInfo str, IndexStmt *node)
 {
-	WRITE_NODE_TYPE("INDEX");
+	WRITE_NODE_TYPE("INDEXSTMT");
 
 	WRITE_STRING_FIELD(idxname);
 	WRITE_NODE_FIELD(relation);
 	WRITE_STRING_FIELD(accessMethod);
+	WRITE_STRING_FIELD(tableSpace);
 	WRITE_NODE_FIELD(indexParams);
 	WRITE_NODE_FIELD(whereClause);
 	WRITE_NODE_FIELD(rangetable);
diff --git a/src/backend/parser/analyze.c b/src/backend/parser/analyze.c
index e5537e26ae9..54501031562 100644
--- a/src/backend/parser/analyze.c
+++ b/src/backend/parser/analyze.c
@@ -6,7 +6,7 @@
  * Portions Copyright (c) 1996-2003, PostgreSQL Global Development Group
  * Portions Copyright (c) 1994, Regents of the University of California
  *
- *	$PostgreSQL: pgsql/src/backend/parser/analyze.c,v 1.305 2004/06/10 17:55:58 tgl Exp $
+ *	$PostgreSQL: pgsql/src/backend/parser/analyze.c,v 1.306 2004/06/18 06:13:31 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -874,6 +874,7 @@ transformColumnDefinition(ParseState *pstate, CreateStmtContext *cxt,
 		seqstmt = makeNode(CreateSeqStmt);
 		seqstmt->sequence = makeRangeVar(snamespace, sname);
 		seqstmt->options = NIL;
+		seqstmt->tablespacename = NULL;
 
 		cxt->blist = lappend(cxt->blist, seqstmt);
 
@@ -1199,6 +1200,7 @@ transformIndexConstraints(ParseState *pstate, CreateStmtContext *cxt)
 
 		index->relation = cxt->relation;
 		index->accessMethod = DEFAULT_INDEX_TYPE;
+		index->tableSpace = NULL;
 		index->indexParams = NIL;
 		index->whereClause = NULL;
 
diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y
index 07a882f3db5..9dc53604c61 100644
--- a/src/backend/parser/gram.y
+++ b/src/backend/parser/gram.y
@@ -11,7 +11,7 @@
  *
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/parser/gram.y,v 2.461 2004/06/09 19:08:17 tgl Exp $
+ *	  $PostgreSQL: pgsql/src/backend/parser/gram.y,v 2.462 2004/06/18 06:13:31 tgl Exp $
  *
  * HISTORY
  *	  AUTHOR			DATE			MAJOR EVENT
@@ -136,12 +136,12 @@ static void doNegateFloat(Value *v);
 		AnalyzeStmt ClosePortalStmt ClusterStmt CommentStmt
 		ConstraintsSetStmt CopyStmt CreateAsStmt CreateCastStmt
 		CreateDomainStmt CreateGroupStmt CreateOpClassStmt CreatePLangStmt
-		CreateSchemaStmt CreateSeqStmt CreateStmt
+		CreateSchemaStmt CreateSeqStmt CreateStmt CreateTableSpaceStmt
 		CreateAssertStmt CreateTrigStmt CreateUserStmt
 		CreatedbStmt DeclareCursorStmt DefineStmt DeleteStmt
 		DropGroupStmt DropOpClassStmt DropPLangStmt DropStmt
 		DropAssertStmt DropTrigStmt DropRuleStmt DropCastStmt
-		DropUserStmt DropdbStmt ExplainStmt FetchStmt
+		DropUserStmt DropdbStmt DropTableSpaceStmt ExplainStmt FetchStmt
 		GrantStmt IndexStmt InsertStmt ListenStmt LoadStmt
 		LockStmt NotifyStmt ExplainableStmt PreparableStmt
 		CreateFunctionStmt ReindexStmt RemoveAggrStmt
@@ -324,6 +324,7 @@ static void doNegateFloat(Value *v);
 
 %type <list>	constraints_set_list
 %type <boolean> constraints_set_mode
+%type <str>		OptTableSpace OptTableSpaceOwner
 
 
 /*
@@ -384,7 +385,7 @@ static void doNegateFloat(Value *v);
 	ORDER OUT_P OUTER_P OVERLAPS OVERLAY OWNER
 
 	PARTIAL PASSWORD PATH_P PENDANT PLACING POSITION
-	PRECISION PRESERVE PREPARE PRIMARY 
+	PRECISION PRESERVE PREPARE PRIMARY
 	PRIOR PRIVILEGES PROCEDURAL PROCEDURE
 
 	QUOTE
@@ -398,7 +399,7 @@ static void doNegateFloat(Value *v);
 	SHOW SIMILAR SIMPLE SMALLINT SOME STABLE START STATEMENT
 	STATISTICS STDIN STDOUT STORAGE STRICT_P SUBSTRING SYSID
 
-	TABLE TEMP TEMPLATE TEMPORARY THEN TIME TIMESTAMP
+	TABLE TABLESPACE TEMP TEMPLATE TEMPORARY THEN TIME TIMESTAMP
 	TO TOAST TRAILING TRANSACTION TREAT TRIGGER TRIM TRUE_P
 	TRUNCATE TRUSTED TYPE_P
 
@@ -513,6 +514,7 @@ stmt :
 			| CreateSchemaStmt
 			| CreateSeqStmt
 			| CreateStmt
+			| CreateTableSpaceStmt
 			| CreateTrigStmt
 			| CreateUserStmt
 			| CreatedbStmt
@@ -527,6 +529,7 @@ stmt :
 			| DropPLangStmt
 			| DropRuleStmt
 			| DropStmt
+			| DropTableSpaceStmt
 			| DropTrigStmt
 			| DropUserStmt
 			| DropdbStmt
@@ -781,7 +784,7 @@ DropGroupStmt:
  *****************************************************************************/
 
 CreateSchemaStmt:
-			CREATE SCHEMA OptSchemaName AUTHORIZATION UserId OptSchemaEltList
+			CREATE SCHEMA OptSchemaName AUTHORIZATION UserId OptTableSpace OptSchemaEltList
 				{
 					CreateSchemaStmt *n = makeNode(CreateSchemaStmt);
 					/* One can omit the schema name or the authorization id. */
@@ -790,16 +793,18 @@ CreateSchemaStmt:
 					else
 						n->schemaname = $5;
 					n->authid = $5;
-					n->schemaElts = $6;
+					n->tablespacename = $6;
+					n->schemaElts = $7;
 					$$ = (Node *)n;
 				}
-			| CREATE SCHEMA ColId OptSchemaEltList
+			| CREATE SCHEMA ColId OptTableSpace OptSchemaEltList
 				{
 					CreateSchemaStmt *n = makeNode(CreateSchemaStmt);
 					/* ...but not both */
 					n->schemaname = $3;
 					n->authid = NULL;
-					n->schemaElts = $4;
+					n->tablespacename = $4;
+					n->schemaElts = $5;
 					$$ = (Node *)n;
 				}
 		;
@@ -1277,7 +1282,7 @@ alter_table_cmd:
 					n->name = $3;
 					$$ = (Node *)n;
 				}
-			/* ALTER TABLE <name> SET WITHOUT CLUSTER */ 
+			/* ALTER TABLE <name> SET WITHOUT CLUSTER */
 			| SET WITHOUT CLUSTER
 				{
 					AlterTableCmd *n = makeNode(AlterTableCmd);
@@ -1464,7 +1469,7 @@ opt_using:
  *****************************************************************************/
 
 CreateStmt:	CREATE OptTemp TABLE qualified_name '(' OptTableElementList ')'
-			OptInherit OptWithOids OnCommitOption
+			OptInherit OptWithOids OnCommitOption OptTableSpace
 				{
 					CreateStmt *n = makeNode(CreateStmt);
 					$4->istemp = $2;
@@ -1474,10 +1479,11 @@ CreateStmt:	CREATE OptTemp TABLE qualified_name '(' OptTableElementList ')'
 					n->constraints = NIL;
 					n->hasoids = $9;
 					n->oncommit = $10;
+					n->tablespacename = $11;
 					$$ = (Node *)n;
 				}
 		| CREATE OptTemp TABLE qualified_name OF qualified_name
-			'(' OptTableElementList ')' OptWithOids OnCommitOption
+			'(' OptTableElementList ')' OptWithOids OnCommitOption OptTableSpace
 				{
 					/* SQL99 CREATE TABLE OF <UDT> (cols) seems to be satisfied
 					 * by our inheritance capabilities. Let's try it...
@@ -1490,6 +1496,7 @@ CreateStmt:	CREATE OptTemp TABLE qualified_name '(' OptTableElementList ')'
 					n->constraints = NIL;
 					n->hasoids = $10;
 					n->oncommit = $11;
+					n->tablespacename = $12;
 					$$ = (Node *)n;
 				}
 		;
@@ -1901,6 +1908,10 @@ OnCommitOption:  ON COMMIT DROP				{ $$ = ONCOMMIT_DROP; }
 			| /*EMPTY*/						{ $$ = ONCOMMIT_NOOP; }
 		;
 
+OptTableSpace:   TABLESPACE name					{ $$ = $2; }
+			| /*EMPTY*/								{ $$ = NULL; }
+		;
+
 
 /*
  * Note: CREATE TABLE ... AS SELECT ... is just another spelling for
@@ -1979,12 +1990,13 @@ CreateAsElement:
  *****************************************************************************/
 
 CreateSeqStmt:
-			CREATE OptTemp SEQUENCE qualified_name OptSeqList
+			CREATE OptTemp SEQUENCE qualified_name OptSeqList OptTableSpace
 				{
 					CreateSeqStmt *n = makeNode(CreateSeqStmt);
 					$4->istemp = $2;
 					n->sequence = $4;
 					n->options = $5;
+					n->tablespacename = $6;
 					$$ = (Node *)n;
 				}
 		;
@@ -2136,6 +2148,45 @@ opt_procedural:
 
 /*****************************************************************************
  *
+ * 		QUERY:
+ *             CREATE TABLESPACE tablespace LOCATION '/path/to/tablespace/'
+ *
+ *****************************************************************************/
+
+CreateTableSpaceStmt: CREATE TABLESPACE name OptTableSpaceOwner LOCATION Sconst
+				{
+					CreateTableSpaceStmt *n = makeNode(CreateTableSpaceStmt);
+					n->tablespacename = $3;
+					n->owner = $4;
+					n->location = $6;
+					$$ = (Node *) n;
+				}
+		;
+
+OptTableSpaceOwner: OWNER name			{ $$ = $2; }
+			| /*EMPTY */				{ $$ = NULL; }
+		;
+
+/*****************************************************************************
+ *
+ * 		QUERY :
+ *				DROP TABLESPACE <tablespace>
+ *
+ *		No need for drop behaviour as we cannot implement dependencies for
+ *		objects in other databases; we can only support RESTRICT.
+ *
+ ****************************************************************************/
+
+DropTableSpaceStmt: DROP TABLESPACE name
+				{
+					DropTableSpaceStmt *n = makeNode(DropTableSpaceStmt);
+					n->tablespacename = $3;
+					$$ = (Node *) n;
+				}
+		;
+
+/*****************************************************************************
+ *
  *		QUERIES :
  *				CREATE TRIGGER ...
  *				DROP TRIGGER ...
@@ -2735,7 +2786,7 @@ CommentStmt:
 					n->objargs = NIL;
 					n->comment = $7;
 					$$ = (Node *) n;
-				}				
+				}
 		;
 
 comment_type:
@@ -3026,6 +3077,13 @@ privilege_target:
 					n->objs = $2;
 					$$ = n;
 				}
+			| TABLESPACE name_list
+				{
+					PrivTarget *n = makeNode(PrivTarget);
+					n->objtype = ACL_OBJECT_TABLESPACE;
+					n->objs = $2;
+					$$ = n;
+				}
 		;
 
 
@@ -3092,12 +3150,14 @@ function_with_argtypes:
  *		QUERY:
  *				create index <indexname> on <relname>
  *				  [ using <access> ] "(" ( <col> [ using <opclass> ] )+ ")"
- *				  [ where <predicate> ]
+ *				  [ tablespace <tablespacename> ] [ where <predicate> ]
  *
+ * Note: we cannot put TABLESPACE clause after WHERE clause unless we are
+ * willing to make TABLESPACE a fully reserved word.
  *****************************************************************************/
 
 IndexStmt:	CREATE index_opt_unique INDEX index_name ON qualified_name
-			access_method_clause '(' index_params ')' where_clause
+			access_method_clause '(' index_params ')' OptTableSpace where_clause
 				{
 					IndexStmt *n = makeNode(IndexStmt);
 					n->unique = $2;
@@ -3105,7 +3165,8 @@ IndexStmt:	CREATE index_opt_unique INDEX index_name ON qualified_name
 					n->relation = $6;
 					n->accessMethod = $7;
 					n->indexParams = $9;
-					n->whereClause = $11;
+					n->tableSpace = $11;
+					n->whereClause = $12;
 					$$ = (Node *)n;
 				}
 		;
@@ -3896,7 +3957,15 @@ createdb_opt_list:
 		;
 
 createdb_opt_item:
-			LOCATION opt_equal Sconst
+			TABLESPACE opt_equal name
+				{
+					$$ = makeDefElem("tablespace", (Node *)makeString($3));
+				}
+			| TABLESPACE opt_equal DEFAULT
+				{
+					$$ = makeDefElem("tablespace", NULL);
+				}
+			| LOCATION opt_equal Sconst
 				{
 					$$ = makeDefElem("location", (Node *)makeString($3));
 				}
@@ -6801,7 +6870,7 @@ subquery_Op:
 					{ $$ = list_make1(makeString("!~~*")); }
 /* cannot put SIMILAR TO here, because SIMILAR TO is a hack.
  * the regular expression is preprocessed by a function (similar_escape),
- * and the ~ operator for posix regular expressions is used. 
+ * and the ~ operator for posix regular expressions is used.
  *        x SIMILAR TO y     ->    x ~ similar_escape(y)
  * this transformation is made on the fly by the parser upwards.
  * however the SubLink structure which handles any/some/all stuff
@@ -6978,7 +7047,7 @@ in_expr:	select_with_parens
  *	COALESCE(a,b,...)
  * same as CASE WHEN a IS NOT NULL THEN a WHEN b IS NOT NULL THEN b ... END
  * - thomas 1998-11-09
- * 
+ *
  * NULLIF and COALESCE have become first class nodes to
  * prevent double evaluation of arguments.
  * - Kris Jurka 2003-02-11
@@ -7565,6 +7634,7 @@ unreserved_keyword:
 			| STORAGE
 			| SYSID
 			| STRICT_P
+			| TABLESPACE
 			| TEMP
 			| TEMPLATE
 			| TEMPORARY
diff --git a/src/backend/parser/keywords.c b/src/backend/parser/keywords.c
index a3c765a1952..a89f8da7d4e 100644
--- a/src/backend/parser/keywords.c
+++ b/src/backend/parser/keywords.c
@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/parser/keywords.c,v 1.149 2004/04/21 00:34:18 momjian Exp $
+ *	  $PostgreSQL: pgsql/src/backend/parser/keywords.c,v 1.150 2004/06/18 06:13:31 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -297,6 +297,7 @@ static const ScanKeyword ScanKeywords[] = {
 	{"substring", SUBSTRING},
 	{"sysid", SYSID},
 	{"table", TABLE},
+	{"tablespace", TABLESPACE},
 	{"temp", TEMP},
 	{"template", TEMPLATE},
 	{"temporary", TEMPORARY},
diff --git a/src/backend/storage/buffer/bufmgr.c b/src/backend/storage/buffer/bufmgr.c
index da1998a0d50..725b79cad38 100644
--- a/src/backend/storage/buffer/bufmgr.c
+++ b/src/backend/storage/buffer/bufmgr.c
@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/storage/buffer/bufmgr.c,v 1.170 2004/06/11 16:43:23 tgl Exp $
+ *	  $PostgreSQL: pgsql/src/backend/storage/buffer/bufmgr.c,v 1.171 2004/06/18 06:13:33 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -654,7 +654,7 @@ BufferSync(int percent, int maxpages)
 	 */
 	dirty_buffers = (BufferDesc **) palloc(NBuffers * sizeof(BufferDesc *));
 	buftags = (BufferTag *) palloc(NBuffers * sizeof(BufferTag));
-	
+
 	LWLockAcquire(BufMgrLock, LW_EXCLUSIVE);
 	num_buffer_dirty = StrategyDirtyBufferList(dirty_buffers, buftags,
 											   NBuffers);
@@ -832,9 +832,10 @@ AtEOXact_Buffers(bool isCommit)
 			if (isCommit)
 				elog(WARNING,
 					 "buffer refcount leak: [%03d] "
-					 "(rel=%u/%u, blockNum=%u, flags=0x%x, refcount=%u %d)",
+					 "(rel=%u/%u/%u, blockNum=%u, flags=0x%x, refcount=%u %d)",
 					 i,
-					 buf->tag.rnode.tblNode, buf->tag.rnode.relNode,
+					 buf->tag.rnode.spcNode, buf->tag.rnode.dbNode,
+					 buf->tag.rnode.relNode,
 					 buf->tag.blockNum, buf->flags,
 					 buf->refcount, PrivateRefCount[i]);
 
@@ -1137,9 +1138,10 @@ recheck:
 			{
 				/* the sole pin should be ours */
 				if (bufHdr->refcount != 1 || PrivateRefCount[i - 1] == 0)
-					elog(FATAL, "block %u of %u/%u is still referenced (private %d, global %u)",
+					elog(FATAL, "block %u of %u/%u/%u is still referenced (private %d, global %u)",
 						 bufHdr->tag.blockNum,
-						 bufHdr->tag.rnode.tblNode,
+						 bufHdr->tag.rnode.spcNode,
+						 bufHdr->tag.rnode.dbNode,
 						 bufHdr->tag.rnode.relNode,
 						 PrivateRefCount[i - 1], bufHdr->refcount);
 				/* Make sure it will be released */
@@ -1180,13 +1182,7 @@ DropBuffers(Oid dbid)
 	{
 		bufHdr = &BufferDescriptors[i - 1];
 recheck:
-
-		/*
-		 * We know that currently database OID is tblNode but this
-		 * probably will be changed in future and this func will be used
-		 * to drop tablespace buffers.
-		 */
-		if (bufHdr->tag.rnode.tblNode == dbid)
+		if (bufHdr->tag.rnode.dbNode == dbid)
 		{
 			/*
 			 * If there is I/O in progress, better wait till it's done;
@@ -1243,10 +1239,11 @@ PrintBufferDescs(void)
 		for (i = 0; i < NBuffers; ++i, ++buf)
 		{
 			elog(LOG,
-				 "[%02d] (freeNext=%d, freePrev=%d, rel=%u/%u, "
+				 "[%02d] (freeNext=%d, freePrev=%d, rel=%u/%u/%u, "
 				 "blockNum=%u, flags=0x%x, refcount=%u %d)",
 				 i, buf->freeNext, buf->freePrev,
-				 buf->tag.rnode.tblNode, buf->tag.rnode.relNode,
+				 buf->tag.rnode.spcNode, buf->tag.rnode.dbNode,
+				 buf->tag.rnode.relNode,
 				 buf->tag.blockNum, buf->flags,
 				 buf->refcount, PrivateRefCount[i]);
 		}
@@ -1257,9 +1254,9 @@ PrintBufferDescs(void)
 		/* interactive backend */
 		for (i = 0; i < NBuffers; ++i, ++buf)
 		{
-			printf("[%-2d] (%u/%u, %u) flags=0x%x, refcount=%u %d)\n",
-				   i, buf->tag.rnode.tblNode, buf->tag.rnode.relNode,
-				   buf->tag.blockNum,
+			printf("[%-2d] (%u/%u/%u, %u) flags=0x%x, refcount=%u %d)\n",
+				   i, buf->tag.rnode.spcNode, buf->tag.rnode.dbNode,
+				   buf->tag.rnode.relNode, buf->tag.blockNum,
 				   buf->flags, buf->refcount, PrivateRefCount[i]);
 		}
 	}
@@ -1278,10 +1275,11 @@ PrintPinnedBufs(void)
 	{
 		if (PrivateRefCount[i] > 0)
 			elog(WARNING,
-				 "[%02d] (freeNext=%d, freePrev=%d, rel=%u/%u, "
+				 "[%02d] (freeNext=%d, freePrev=%d, rel=%u/%u/%u, "
 				 "blockNum=%u, flags=0x%x, refcount=%u %d)",
 				 i, buf->freeNext, buf->freePrev,
-				 buf->tag.rnode.tblNode, buf->tag.rnode.relNode,
+				 buf->tag.rnode.spcNode, buf->tag.rnode.dbNode,
+				 buf->tag.rnode.relNode,
 				 buf->tag.blockNum, buf->flags,
 				 buf->refcount, PrivateRefCount[i]);
 	}
@@ -1464,11 +1462,11 @@ IncrBufferRefCount_Debug(char *file, int line, Buffer buffer)
 		BufferDesc *buf = &BufferDescriptors[buffer - 1];
 
 		fprintf(stderr,
-				"PIN(Incr) %d rel = %u/%u, blockNum = %u, "
+				"PIN(Incr) %d rel = %u/%u/%u, blockNum = %u, "
 				"refcount = %d, file: %s, line: %d\n",
 				buffer,
-				buf->tag.rnode.tblNode, buf->tag.rnode.relNode,
-				buf->tag.blockNum,
+				buf->tag.rnode.spcNode, buf->tag.rnode.dbNode,
+				buf->tag.rnode.relNode, buf->tag.blockNum,
 				PrivateRefCount[buffer - 1], file, line);
 	}
 }
@@ -1484,11 +1482,11 @@ ReleaseBuffer_Debug(char *file, int line, Buffer buffer)
 		BufferDesc *buf = &BufferDescriptors[buffer - 1];
 
 		fprintf(stderr,
-				"UNPIN(Rel) %d rel = %u/%u, blockNum = %u, "
+				"UNPIN(Rel) %d rel = %u/%u/%u, blockNum = %u, "
 				"refcount = %d, file: %s, line: %d\n",
 				buffer,
-				buf->tag.rnode.tblNode, buf->tag.rnode.relNode,
-				buf->tag.blockNum,
+				buf->tag.rnode.spcNode, buf->tag.rnode.dbNode,
+				buf->tag.rnode.relNode, buf->tag.blockNum,
 				PrivateRefCount[buffer - 1], file, line);
 	}
 }
@@ -1513,11 +1511,11 @@ ReleaseAndReadBuffer_Debug(char *file,
 		BufferDesc *buf = &BufferDescriptors[buffer - 1];
 
 		fprintf(stderr,
-				"UNPIN(Rel&Rd) %d rel = %u/%u, blockNum = %u, "
+				"UNPIN(Rel&Rd) %d rel = %u/%u/%u, blockNum = %u, "
 				"refcount = %d, file: %s, line: %d\n",
 				buffer,
-				buf->tag.rnode.tblNode, buf->tag.rnode.relNode,
-				buf->tag.blockNum,
+				buf->tag.rnode.spcNode, buf->tag.rnode.dbNode,
+				buf->tag.rnode.relNode, buf->tag.blockNum,
 				PrivateRefCount[buffer - 1], file, line);
 	}
 	if (ShowPinTrace && BufferIsLocal(buffer) && is_userbuffer(buffer))
@@ -1525,11 +1523,11 @@ ReleaseAndReadBuffer_Debug(char *file,
 		BufferDesc *buf = &BufferDescriptors[b - 1];
 
 		fprintf(stderr,
-				"PIN(Rel&Rd) %d rel = %u/%u, blockNum = %u, "
+				"PIN(Rel&Rd) %d rel = %u/%u/%u, blockNum = %u, "
 				"refcount = %d, file: %s, line: %d\n",
 				b,
-				buf->tag.rnode.tblNode, buf->tag.rnode.relNode,
-				buf->tag.blockNum,
+				buf->tag.rnode.spcNode, buf->tag.rnode.dbNode,
+				buf->tag.rnode.relNode, buf->tag.blockNum,
 				PrivateRefCount[b - 1], file, line);
 	}
 	return b;
@@ -1890,9 +1888,10 @@ AbortBufferIO(void)
 			{
 				ereport(WARNING,
 						(errcode(ERRCODE_IO_ERROR),
-						 errmsg("could not write block %u of %u/%u",
+						 errmsg("could not write block %u of %u/%u/%u",
 								buf->tag.blockNum,
-								buf->tag.rnode.tblNode,
+								buf->tag.rnode.spcNode,
+								buf->tag.rnode.dbNode,
 								buf->tag.rnode.relNode),
 						 errdetail("Multiple failures --- write error may be permanent.")));
 			}
@@ -1912,7 +1911,9 @@ buffer_write_error_callback(void *arg)
 	BufferDesc *bufHdr = (BufferDesc *) arg;
 
 	if (bufHdr != NULL)
-		errcontext("writing block %u of relation %u/%u",
+		errcontext("writing block %u of relation %u/%u/%u",
 				   bufHdr->tag.blockNum,
-				   bufHdr->tag.rnode.tblNode, bufHdr->tag.rnode.relNode);
+				   bufHdr->tag.rnode.spcNode,
+				   bufHdr->tag.rnode.dbNode,
+				   bufHdr->tag.rnode.relNode);
 }
diff --git a/src/backend/storage/buffer/localbuf.c b/src/backend/storage/buffer/localbuf.c
index 95e86e955d1..f4d1163f16a 100644
--- a/src/backend/storage/buffer/localbuf.c
+++ b/src/backend/storage/buffer/localbuf.c
@@ -9,7 +9,7 @@
  *
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/storage/buffer/localbuf.c,v 1.55 2004/05/31 20:31:33 tgl Exp $
+ *	  $PostgreSQL: pgsql/src/backend/storage/buffer/localbuf.c,v 1.56 2004/06/18 06:13:33 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -236,10 +236,10 @@ AtEOXact_LocalBuffers(bool isCommit)
 
 			if (isCommit)
 				elog(WARNING,
-					 "local buffer leak: [%03d] (rel=%u/%u, blockNum=%u, flags=0x%x, refcount=%u %d)",
+					 "local buffer leak: [%03d] (rel=%u/%u/%u, blockNum=%u, flags=0x%x, refcount=%u %d)",
 					 i,
-					 buf->tag.rnode.tblNode, buf->tag.rnode.relNode,
-					 buf->tag.blockNum, buf->flags,
+					 buf->tag.rnode.spcNode, buf->tag.rnode.dbNode,
+					 buf->tag.rnode.relNode, buf->tag.blockNum, buf->flags,
 					 buf->refcount, LocalRefCount[i]);
 
 			LocalRefCount[i] = 0;
diff --git a/src/backend/storage/freespace/freespace.c b/src/backend/storage/freespace/freespace.c
index 527ad8496ab..28815e6d77f 100644
--- a/src/backend/storage/freespace/freespace.c
+++ b/src/backend/storage/freespace/freespace.c
@@ -8,7 +8,7 @@
  * Portions Copyright (c) 1994, Regents of the University of California
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/storage/freespace/freespace.c,v 1.31 2004/06/05 19:48:08 tgl Exp $
+ *	  $PostgreSQL: pgsql/src/backend/storage/freespace/freespace.c,v 1.32 2004/06/18 06:13:34 tgl Exp $
  *
  *
  * NOTES:
@@ -658,9 +658,6 @@ FreeSpaceMapForgetRel(RelFileNode *rel)
  *
  * This is called during DROP DATABASE.  As above, might as well reclaim
  * map space sooner instead of later.
- *
- * XXX when we implement tablespaces, target Oid will need to be tablespace
- * ID not database ID.
  */
 void
 FreeSpaceMapForgetDatabase(Oid dbid)
@@ -672,7 +669,7 @@ FreeSpaceMapForgetDatabase(Oid dbid)
 	for (fsmrel = FreeSpaceMap->usageList; fsmrel; fsmrel = nextrel)
 	{
 		nextrel = fsmrel->nextUsage;	/* in case we delete it */
-		if (fsmrel->key.tblNode == dbid)
+		if (fsmrel->key.dbNode == dbid)
 			delete_fsm_rel(fsmrel);
 	}
 	LWLockRelease(FreeSpaceLock);
@@ -1847,8 +1844,9 @@ DumpFreeSpace(void)
 	for (fsmrel = FreeSpaceMap->usageList; fsmrel; fsmrel = fsmrel->nextUsage)
 	{
 		relNum++;
-		fprintf(stderr, "Map %d: rel %u/%u isIndex %d avgRequest %u lastPageCount %d nextPage %d\nMap= ",
-				relNum, fsmrel->key.tblNode, fsmrel->key.relNode,
+		fprintf(stderr, "Map %d: rel %u/%u/%u isIndex %d avgRequest %u lastPageCount %d nextPage %d\nMap= ",
+				relNum,
+				fsmrel->key.spcNode, fsmrel->key.dbNode, fsmrel->key.relNode,
 				(int) fsmrel->isIndex, fsmrel->avgRequest,
 				fsmrel->lastPageCount, fsmrel->nextPage);
 		if (fsmrel->isIndex)
diff --git a/src/backend/storage/smgr/md.c b/src/backend/storage/smgr/md.c
index 4f0d241215d..b0d667b8369 100644
--- a/src/backend/storage/smgr/md.c
+++ b/src/backend/storage/smgr/md.c
@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/storage/smgr/md.c,v 1.107 2004/06/02 17:28:18 tgl Exp $
+ *	  $PostgreSQL: pgsql/src/backend/storage/smgr/md.c,v 1.108 2004/06/18 06:13:37 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -363,8 +363,9 @@ mdopen(SMgrRelation reln, bool allowNotFound)
 				return NULL;
 			ereport(ERROR,
 					(errcode_for_file_access(),
-					 errmsg("could not open relation %u/%u: %m",
-							reln->smgr_rnode.tblNode,
+					 errmsg("could not open relation %u/%u/%u: %m",
+							reln->smgr_rnode.spcNode,
+							reln->smgr_rnode.dbNode,
 							reln->smgr_rnode.relNode)));
 		}
 	}
@@ -765,9 +766,10 @@ mdsync(void)
 				{
 					ereport(LOG,
 							(errcode_for_file_access(),
-							 errmsg("could not fsync segment %u of relation %u/%u: %m",
+							 errmsg("could not fsync segment %u of relation %u/%u/%u: %m",
 									entry->segno,
-									entry->rnode.tblNode,
+									entry->rnode.spcNode,
+									entry->rnode.dbNode,
 									entry->rnode.relNode)));
 					return false;
 				}
@@ -945,9 +947,10 @@ _mdfd_getseg(SMgrRelation reln, BlockNumber blkno, bool allowNotFound)
 					return NULL;
 				ereport(ERROR,
 						(errcode_for_file_access(),
-						 errmsg("could not open segment %u of relation %u/%u (target block %u): %m",
+						 errmsg("could not open segment %u of relation %u/%u/%u (target block %u): %m",
 								nextsegno,
-								reln->smgr_rnode.tblNode,
+								reln->smgr_rnode.spcNode,
+								reln->smgr_rnode.dbNode,
 								reln->smgr_rnode.relNode,
 								blkno)));
 			}
diff --git a/src/backend/storage/smgr/smgr.c b/src/backend/storage/smgr/smgr.c
index 8977f026e4f..c7783d878f2 100644
--- a/src/backend/storage/smgr/smgr.c
+++ b/src/backend/storage/smgr/smgr.c
@@ -11,7 +11,7 @@
  *
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/storage/smgr/smgr.c,v 1.73 2004/06/02 17:28:18 tgl Exp $
+ *	  $PostgreSQL: pgsql/src/backend/storage/smgr/smgr.c,v 1.74 2004/06/18 06:13:37 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -227,8 +227,9 @@ smgrclose(SMgrRelation reln)
 	if (! (*(smgrsw[reln->smgr_which].smgr_close)) (reln))
 		ereport(ERROR,
 				(errcode_for_file_access(),
-				 errmsg("could not close relation %u/%u: %m",
-						reln->smgr_rnode.tblNode,
+				 errmsg("could not close relation %u/%u/%u: %m",
+						reln->smgr_rnode.spcNode,
+						reln->smgr_rnode.dbNode,
 						reln->smgr_rnode.relNode)));
 
 	if (hash_search(SMgrRelationHash,
@@ -308,8 +309,9 @@ smgrcreate(SMgrRelation reln, bool isTemp, bool isRedo)
 	if (! (*(smgrsw[reln->smgr_which].smgr_create)) (reln, isRedo))
 		ereport(ERROR,
 				(errcode_for_file_access(),
-				 errmsg("could not create relation %u/%u: %m",
-						reln->smgr_rnode.tblNode,
+				 errmsg("could not create relation %u/%u/%u: %m",
+						reln->smgr_rnode.spcNode,
+						reln->smgr_rnode.dbNode,
 						reln->smgr_rnode.relNode)));
 
 	if (isRedo)
@@ -427,8 +429,9 @@ smgr_internal_unlink(RelFileNode rnode, int which, bool isTemp, bool isRedo)
 	if (! (*(smgrsw[which].smgr_unlink)) (rnode, isRedo))
 		ereport(WARNING,
 				(errcode_for_file_access(),
-				 errmsg("could not unlink relation %u/%u: %m",
-						rnode.tblNode,
+				 errmsg("could not unlink relation %u/%u/%u: %m",
+						rnode.spcNode,
+						rnode.dbNode,
 						rnode.relNode)));
 }
 
@@ -447,8 +450,9 @@ smgrextend(SMgrRelation reln, BlockNumber blocknum, char *buffer, bool isTemp)
 													 isTemp))
 		ereport(ERROR,
 				(errcode_for_file_access(),
-				 errmsg("could not extend relation %u/%u: %m",
-						reln->smgr_rnode.tblNode,
+				 errmsg("could not extend relation %u/%u/%u: %m",
+						reln->smgr_rnode.spcNode,
+						reln->smgr_rnode.dbNode,
 						reln->smgr_rnode.relNode),
 				 errhint("Check free disk space.")));
 }
@@ -467,9 +471,10 @@ smgrread(SMgrRelation reln, BlockNumber blocknum, char *buffer)
 	if (! (*(smgrsw[reln->smgr_which].smgr_read)) (reln, blocknum, buffer))
 		ereport(ERROR,
 				(errcode_for_file_access(),
-				 errmsg("could not read block %u of relation %u/%u: %m",
+				 errmsg("could not read block %u of relation %u/%u/%u: %m",
 						blocknum,
-						reln->smgr_rnode.tblNode,
+						reln->smgr_rnode.spcNode,
+						reln->smgr_rnode.dbNode,
 						reln->smgr_rnode.relNode)));
 }
 
@@ -491,9 +496,10 @@ smgrwrite(SMgrRelation reln, BlockNumber blocknum, char *buffer, bool isTemp)
 													isTemp))
 		ereport(ERROR,
 				(errcode_for_file_access(),
-				 errmsg("could not write block %u of relation %u/%u: %m",
+				 errmsg("could not write block %u of relation %u/%u/%u: %m",
 						blocknum,
-						reln->smgr_rnode.tblNode,
+						reln->smgr_rnode.spcNode,
+						reln->smgr_rnode.dbNode,
 						reln->smgr_rnode.relNode)));
 }
 
@@ -520,8 +526,9 @@ smgrnblocks(SMgrRelation reln)
 	if (nblocks == InvalidBlockNumber)
 		ereport(ERROR,
 				(errcode_for_file_access(),
-				 errmsg("could not count blocks of relation %u/%u: %m",
-						reln->smgr_rnode.tblNode,
+				 errmsg("could not count blocks of relation %u/%u/%u: %m",
+						reln->smgr_rnode.spcNode,
+						reln->smgr_rnode.dbNode,
 						reln->smgr_rnode.relNode)));
 
 	return nblocks;
@@ -552,8 +559,9 @@ smgrtruncate(SMgrRelation reln, BlockNumber nblocks, bool isTemp)
 	if (newblks == InvalidBlockNumber)
 		ereport(ERROR,
 				(errcode_for_file_access(),
-				 errmsg("could not truncate relation %u/%u to %u blocks: %m",
-						reln->smgr_rnode.tblNode,
+				 errmsg("could not truncate relation %u/%u/%u to %u blocks: %m",
+						reln->smgr_rnode.spcNode,
+						reln->smgr_rnode.dbNode,
 						reln->smgr_rnode.relNode,
 						nblocks)));
 
@@ -607,8 +615,9 @@ smgrimmedsync(SMgrRelation reln)
 	if (! (*(smgrsw[reln->smgr_which].smgr_immedsync)) (reln))
 		ereport(ERROR,
 				(errcode_for_file_access(),
-				 errmsg("could not sync relation %u/%u: %m",
-						reln->smgr_rnode.tblNode,
+				 errmsg("could not sync relation %u/%u/%u: %m",
+						reln->smgr_rnode.spcNode,
+						reln->smgr_rnode.dbNode,
 						reln->smgr_rnode.relNode)));
 }
 
@@ -775,8 +784,9 @@ smgr_redo(XLogRecPtr lsn, XLogRecord *record)
 		if (newblks == InvalidBlockNumber)
 			ereport(WARNING,
 					(errcode_for_file_access(),
-					 errmsg("could not truncate relation %u/%u to %u blocks: %m",
-							reln->smgr_rnode.tblNode,
+					 errmsg("could not truncate relation %u/%u/%u to %u blocks: %m",
+							reln->smgr_rnode.spcNode,
+							reln->smgr_rnode.dbNode,
 							reln->smgr_rnode.relNode,
 							xlrec->blkno)));
 	}
@@ -800,16 +810,17 @@ smgr_desc(char *buf, uint8 xl_info, char *rec)
 	{
 		xl_smgr_create *xlrec = (xl_smgr_create *) rec;
 
-		sprintf(buf + strlen(buf), "file create: %u/%u",
-				xlrec->rnode.tblNode, xlrec->rnode.relNode);
+		sprintf(buf + strlen(buf), "file create: %u/%u/%u",
+				xlrec->rnode.spcNode, xlrec->rnode.dbNode,
+				xlrec->rnode.relNode);
 	}
 	else if (info == XLOG_SMGR_TRUNCATE)
 	{
 		xl_smgr_truncate *xlrec = (xl_smgr_truncate *) rec;
 
-		sprintf(buf + strlen(buf), "file truncate: %u/%u to %u blocks",
-				xlrec->rnode.tblNode, xlrec->rnode.relNode,
-				xlrec->blkno);
+		sprintf(buf + strlen(buf), "file truncate: %u/%u/%u to %u blocks",
+				xlrec->rnode.spcNode, xlrec->rnode.dbNode,
+				xlrec->rnode.relNode, xlrec->blkno);
 	}
 	else
 		strcat(buf, "UNKNOWN");
diff --git a/src/backend/tcop/utility.c b/src/backend/tcop/utility.c
index 4d6c246cea1..d12cf0d750f 100644
--- a/src/backend/tcop/utility.c
+++ b/src/backend/tcop/utility.c
@@ -10,7 +10,7 @@
  *
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/tcop/utility.c,v 1.218 2004/05/29 22:48:20 tgl Exp $
+ *	  $PostgreSQL: pgsql/src/backend/tcop/utility.c,v 1.219 2004/06/18 06:13:38 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -36,6 +36,7 @@
 #include "commands/schemacmds.h"
 #include "commands/sequence.h"
 #include "commands/tablecmds.h"
+#include "commands/tablespace.h"
 #include "commands/trigger.h"
 #include "commands/typecmds.h"
 #include "commands/user.h"
@@ -258,6 +259,7 @@ check_xact_readonly(Node *parsetree)
 		case T_CreateSchemaStmt:
 		case T_CreateSeqStmt:
 		case T_CreateStmt:
+		case T_CreateTableSpaceStmt:
 		case T_CreateTrigStmt:
 		case T_CompositeTypeStmt:
 		case T_CreateUserStmt:
@@ -266,6 +268,7 @@ check_xact_readonly(Node *parsetree)
 		case T_DropCastStmt:
 		case T_DropStmt:
 		case T_DropdbStmt:
+		case T_DropTableSpaceStmt:
 		case T_RemoveFuncStmt:
 		case T_DropGroupStmt:
 		case T_DropPLangStmt:
@@ -404,6 +407,14 @@ ProcessUtility(Node *parsetree,
 			}
 			break;
 
+		case T_CreateTableSpaceStmt:
+			CreateTableSpace((CreateTableSpaceStmt *) parsetree);
+			break;
+
+		case T_DropTableSpaceStmt:
+			DropTableSpace((DropTableSpaceStmt *) parsetree);
+			break;
+
 		case T_DropStmt:
 			{
 				DropStmt   *stmt = (DropStmt *) parsetree;
@@ -636,6 +647,7 @@ ProcessUtility(Node *parsetree,
 				DefineIndex(stmt->relation,		/* relation */
 							stmt->idxname,		/* index name */
 							stmt->accessMethod, /* am name */
+							stmt->tableSpace,
 							stmt->indexParams,	/* parameters */
 							(Expr *) stmt->whereClause,
 							stmt->rangetable,
@@ -1153,6 +1165,14 @@ CreateCommandTag(Node *parsetree)
 			tag = "CREATE TABLE";
 			break;
 
+		case T_CreateTableSpaceStmt:
+			tag = "CREATE TABLESPACE";
+			break;
+
+		case T_DropTableSpaceStmt:
+			tag = "DROP TABLESPACE";
+			break;
+
 		case T_DropStmt:
 			switch (((DropStmt *) parsetree)->removeType)
 			{
@@ -1224,6 +1244,9 @@ CreateCommandTag(Node *parsetree)
 				case OBJECT_SCHEMA:
 					tag = "ALTER SCHEMA";
 					break;
+				case OBJECT_TABLESPACE:
+					tag = "ALTER TABLESPACE";
+					break;
 				case OBJECT_TRIGGER:
 					tag = "ALTER TRIGGER";
 					break;
diff --git a/src/backend/utils/adt/acl.c b/src/backend/utils/adt/acl.c
index d02683245ac..4b13e318be6 100644
--- a/src/backend/utils/adt/acl.c
+++ b/src/backend/utils/adt/acl.c
@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/utils/adt/acl.c,v 1.105 2004/06/01 21:49:22 tgl Exp $
+ *	  $PostgreSQL: pgsql/src/backend/utils/adt/acl.c,v 1.106 2004/06/18 06:13:49 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -539,6 +539,10 @@ acldefault(GrantObjectType objtype, AclId ownerid)
 			world_default = ACL_NO_RIGHTS;
 			owner_default = ACL_ALL_RIGHTS_NAMESPACE;
 			break;
+		case ACL_OBJECT_TABLESPACE:
+			world_default = ACL_NO_RIGHTS;
+			owner_default = ACL_ALL_RIGHTS_TABLESPACE;
+			break;
 		default:
 			elog(ERROR, "unrecognized objtype: %d", (int) objtype);
 			world_default = ACL_NO_RIGHTS;		/* keep compiler quiet */
diff --git a/src/backend/utils/adt/ruleutils.c b/src/backend/utils/adt/ruleutils.c
index 4caaabd62cd..1a2fe54d7ac 100644
--- a/src/backend/utils/adt/ruleutils.c
+++ b/src/backend/utils/adt/ruleutils.c
@@ -3,7 +3,7 @@
  *				back to source text
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/utils/adt/ruleutils.c,v 1.172 2004/06/16 01:26:47 tgl Exp $
+ *	  $PostgreSQL: pgsql/src/backend/utils/adt/ruleutils.c,v 1.173 2004/06/18 06:13:49 tgl Exp $
  *
  *	  This software is copyrighted by Jan Wieck - Hamburg.
  *
@@ -53,6 +53,7 @@
 #include "catalog/pg_operator.h"
 #include "catalog/pg_shadow.h"
 #include "catalog/pg_trigger.h"
+#include "commands/tablespace.h"
 #include "executor/spi.h"
 #include "lib/stringinfo.h"
 #include "nodes/makefuncs.h"
@@ -768,6 +769,23 @@ pg_get_indexdef_worker(Oid indexrelid, int colno, int prettyFlags)
 		appendStringInfoChar(&buf, ')');
 
 		/*
+		 * If the index is in a different tablespace from its parent,
+		 * tell about that
+		 */
+		if (OidIsValid(idxrelrec->reltablespace) &&
+			idxrelrec->reltablespace != get_rel_tablespace(indrelid))
+		{
+			char	*spcname = get_tablespace_name(idxrelrec->reltablespace);
+
+			if (spcname)		/* just paranoia... */
+			{
+				appendStringInfo(&buf, " TABLESPACE %s",
+								 quote_identifier(spcname));
+				pfree(spcname);
+			}
+		}
+
+		/*
 		 * If it's a partial index, decompile and append the predicate
 		 */
 		if (!heap_attisnull(ht_idx, Anum_pg_index_indpred))
diff --git a/src/backend/utils/cache/inval.c b/src/backend/utils/cache/inval.c
index 7b00f0531e2..ea958a27b46 100644
--- a/src/backend/utils/cache/inval.c
+++ b/src/backend/utils/cache/inval.c
@@ -74,7 +74,7 @@
  * Portions Copyright (c) 1994, Regents of the University of California
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/utils/cache/inval.c,v 1.61 2004/05/06 16:10:57 tgl Exp $
+ *	  $PostgreSQL: pgsql/src/backend/utils/cache/inval.c,v 1.62 2004/06/18 06:13:52 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -308,7 +308,7 @@ AddRelcacheInvalidationMessage(InvalidationListHeader *hdr,
 	/* We assume dbId need not be checked because it will never change */
 	/* relfilenode fields must be checked to support reassignment */
 	ProcessMessageList(hdr->rclist,
-					   if (msg->rc.relId == relId && 
+					   if (msg->rc.relId == relId &&
 						   RelFileNodeEquals(msg->rc.physId, physId)) return);
 
 	/* OK, add the item */
@@ -555,14 +555,18 @@ PrepareForTupleInvalidation(Relation relation, HeapTuple tuple,
 			databaseId = InvalidOid;
 		else
 			databaseId = MyDatabaseId;
-		rnode.tblNode = databaseId;			/* XXX change for tablespaces */
+		if (classtup->reltablespace)
+			rnode.spcNode = classtup->reltablespace;
+		else
+			rnode.spcNode = MyDatabaseTableSpace;
+		rnode.dbNode = databaseId;
 		rnode.relNode = classtup->relfilenode;
 		/*
 		 * Note: during a pg_class row update that assigns a new relfilenode
-		 * value, we will be called on both the old and new tuples, and thus
-		 * will broadcast invalidation messages showing both the old and new
-		 * relfilenode values.  This ensures that other backends will close
-		 * smgr references to the old relfilenode file.
+		 * or reltablespace value, we will be called on both the old and new
+		 * tuples, and thus will broadcast invalidation messages showing both
+		 * the old and new RelFileNode values.  This ensures that other
+		 * backends will close smgr references to the old file.
 		 */
 	}
 	else if (tupleRelId == RelOid_pg_attribute)
@@ -580,7 +584,8 @@ PrepareForTupleInvalidation(Relation relation, HeapTuple tuple,
 		 */
 		databaseId = MyDatabaseId;
 		/* We assume no smgr cache flush is needed, either */
-		rnode.tblNode = InvalidOid;
+		rnode.spcNode = InvalidOid;
+		rnode.dbNode = InvalidOid;
 		rnode.relNode = InvalidOid;
 	}
 	else
@@ -760,7 +765,11 @@ CacheInvalidateRelcacheByTuple(HeapTuple classTuple)
 		databaseId = InvalidOid;
 	else
 		databaseId = MyDatabaseId;
-	rnode.tblNode = databaseId;			/* XXX change for tablespaces */
+	if (classtup->reltablespace)
+		rnode.spcNode = classtup->reltablespace;
+	else
+		rnode.spcNode = MyDatabaseTableSpace;
+	rnode.dbNode = databaseId;
 	rnode.relNode = classtup->relfilenode;
 
 	RegisterRelcacheInvalidation(databaseId, relationId, rnode);
diff --git a/src/backend/utils/cache/lsyscache.c b/src/backend/utils/cache/lsyscache.c
index d51d1c18925..1621982502a 100644
--- a/src/backend/utils/cache/lsyscache.c
+++ b/src/backend/utils/cache/lsyscache.c
@@ -7,7 +7,7 @@
  * Portions Copyright (c) 1994, Regents of the University of California
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/utils/cache/lsyscache.c,v 1.113 2004/06/06 00:41:27 tgl Exp $
+ *	  $PostgreSQL: pgsql/src/backend/utils/cache/lsyscache.c,v 1.114 2004/06/18 06:13:52 tgl Exp $
  *
  * NOTES
  *	  Eventually, the index information should go through here, too.
@@ -986,6 +986,34 @@ get_rel_namespace(Oid relid)
 }
 
 /*
+ * get_rel_tablespace
+ *		Returns the pg_tablespace OID associated with a given relation.
+ *
+ * Note: failure return is InvalidOid, which cannot be distinguished from
+ * "default tablespace for this database", but that seems OK.
+ */
+Oid
+get_rel_tablespace(Oid relid)
+{
+	HeapTuple	tp;
+
+	tp = SearchSysCache(RELOID,
+						ObjectIdGetDatum(relid),
+						0, 0, 0);
+	if (HeapTupleIsValid(tp))
+	{
+		Form_pg_class reltup = (Form_pg_class) GETSTRUCT(tp);
+		Oid			result;
+
+		result = reltup->reltablespace;
+		ReleaseSysCache(tp);
+		return result;
+	}
+	else
+		return InvalidOid;
+}
+
+/*
  * get_rel_type_id
  *
  *		Returns the pg_type OID associated with a given relation.
@@ -1980,6 +2008,34 @@ get_namespace_name(Oid nspid)
 		return NULL;
 }
 
+/*
+ * get_namespace_tablespace
+ *		Returns the default tablespace of a given namespace
+ *
+ * Note: failure return is InvalidOid, which cannot be distinguished from
+ * "default tablespace for this database", but that seems OK.
+ */
+Oid
+get_namespace_tablespace(Oid nspid)
+{
+	HeapTuple	tp;
+
+	tp = SearchSysCache(NAMESPACEOID,
+						ObjectIdGetDatum(nspid),
+						0, 0, 0);
+	if (HeapTupleIsValid(tp))
+	{
+		Form_pg_namespace nsptup = (Form_pg_namespace) GETSTRUCT(tp);
+		Oid			result;
+
+		result = nsptup->nsptablespace;
+		ReleaseSysCache(tp);
+		return result;
+	}
+	else
+		return InvalidOid;
+}
+
 /*				---------- PG_SHADOW CACHE ----------					 */
 
 /*
diff --git a/src/backend/utils/cache/relcache.c b/src/backend/utils/cache/relcache.c
index 4221976f0b1..ee8b46407e1 100644
--- a/src/backend/utils/cache/relcache.c
+++ b/src/backend/utils/cache/relcache.c
@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/utils/cache/relcache.c,v 1.204 2004/05/30 23:40:37 neilc Exp $
+ *	  $PostgreSQL: pgsql/src/backend/utils/cache/relcache.c,v 1.205 2004/06/18 06:13:52 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -260,6 +260,7 @@ static void RelationBuildTupleDesc(RelationBuildDescInfo buildinfo,
 					   Relation relation);
 static Relation RelationBuildDesc(RelationBuildDescInfo buildinfo,
 				  Relation oldrelation);
+static void RelationInitPhysicalAddr(Relation relation);
 static void AttrDefaultFetch(Relation relation);
 static void CheckConstraintFetch(Relation relation);
 static List *insert_ordered_oid(List *list, Oid datum);
@@ -873,11 +874,10 @@ RelationBuildDesc(RelationBuildDescInfo buildinfo,
 	 */
 	RelationInitLockInfo(relation);		/* see lmgr.c */
 
-	if (relation->rd_rel->relisshared)
-		relation->rd_node.tblNode = InvalidOid;
-	else
-		relation->rd_node.tblNode = MyDatabaseId;
-	relation->rd_node.relNode = relation->rd_rel->relfilenode;
+	/*
+	 * initialize physical addressing information for the relation
+	 */
+	RelationInitPhysicalAddr(relation);
 
 	/* make sure relation is marked as having no open file yet */
 	relation->rd_smgr = NULL;
@@ -893,6 +893,23 @@ RelationBuildDesc(RelationBuildDescInfo buildinfo,
 }
 
 /*
+ * Initialize the physical addressing info (RelFileNode) for a relcache entry
+ */
+static void
+RelationInitPhysicalAddr(Relation relation)
+{
+	if (relation->rd_rel->reltablespace)
+		relation->rd_node.spcNode = relation->rd_rel->reltablespace;
+	else
+		relation->rd_node.spcNode = MyDatabaseTableSpace;
+	if (relation->rd_rel->relisshared)
+		relation->rd_node.dbNode = InvalidOid;
+	else
+		relation->rd_node.dbNode = MyDatabaseId;
+	relation->rd_node.relNode = relation->rd_rel->relfilenode;
+}
+
+/*
  * Initialize index-access-method support data for an index relation
  */
 void
@@ -1343,18 +1360,17 @@ formrdesc(const char *relationName,
 	 * initialize relation id from info in att array (my, this is ugly)
 	 */
 	RelationGetRelid(relation) = relation->rd_att->attrs[0]->attrelid;
+	relation->rd_rel->relfilenode = RelationGetRelid(relation);
 
 	/*
-	 * initialize the relation's lock manager and RelFileNode information
+	 * initialize the relation lock manager information
 	 */
 	RelationInitLockInfo(relation);		/* see lmgr.c */
 
-	if (relation->rd_rel->relisshared)
-		relation->rd_node.tblNode = InvalidOid;
-	else
-		relation->rd_node.tblNode = MyDatabaseId;
-	relation->rd_node.relNode =
-		relation->rd_rel->relfilenode = RelationGetRelid(relation);
+	/*
+	 * initialize physical addressing information for the relation
+	 */
+	RelationInitPhysicalAddr(relation);
 
 	/*
 	 * initialize the rel-has-index flag, using hardwired knowledge
@@ -1570,7 +1586,8 @@ RelationReloadClassinfo(Relation relation)
 			 relation->rd_id);
 	relp = (Form_pg_class) GETSTRUCT(pg_class_tuple);
 	memcpy((char *) relation->rd_rel, (char *) relp, CLASS_TUPLE_SIZE);
-	relation->rd_node.relNode = relp->relfilenode;
+	/* Now we can recalculate physical address */
+	RelationInitPhysicalAddr(relation);
 	heap_freetuple(pg_class_tuple);
 	relation->rd_targblock = InvalidBlockNumber;
 	/* Okay, now it's valid again */
@@ -2040,8 +2057,9 @@ Relation
 RelationBuildLocalRelation(const char *relname,
 						   Oid relnamespace,
 						   TupleDesc tupDesc,
-						   Oid relid, Oid dbid,
-						   RelFileNode rnode,
+						   Oid relid,
+						   Oid reltablespace,
+						   bool shared_relation,
 						   bool nailit)
 {
 	Relation	rel;
@@ -2125,20 +2143,23 @@ RelationBuildLocalRelation(const char *relname,
 
 	/*
 	 * Insert relation physical and logical identifiers (OIDs) into the
-	 * right places.
+	 * right places.  Note that the physical ID (relfilenode) is initially
+	 * the same as the logical ID (OID).
 	 */
-	rel->rd_rel->relisshared = (dbid == InvalidOid);
+	rel->rd_rel->relisshared = shared_relation;
 
 	RelationGetRelid(rel) = relid;
 
 	for (i = 0; i < natts; i++)
 		rel->rd_att->attrs[i]->attrelid = relid;
 
-	rel->rd_node = rnode;
-	rel->rd_rel->relfilenode = rnode.relNode;
+	rel->rd_rel->relfilenode = relid;
+	rel->rd_rel->reltablespace = reltablespace;
 
 	RelationInitLockInfo(rel);	/* see lmgr.c */
 
+	RelationInitPhysicalAddr(rel);
+
 	/*
 	 * Okay to insert into the relcache hash tables.
 	 */
@@ -3053,16 +3074,12 @@ load_relcache_init_file(void)
 		MemSet(&rel->pgstat_info, 0, sizeof(rel->pgstat_info));
 
 		/*
-		 * Make sure database ID is correct.  This is needed in case the
-		 * pg_internal.init file was copied from some other database by
-		 * CREATE DATABASE.
+		 * Recompute lock and physical addressing info.  This is needed in
+		 * case the pg_internal.init file was copied from some other database
+		 * by CREATE DATABASE.
 		 */
-		if (rel->rd_rel->relisshared)
-			rel->rd_node.tblNode = InvalidOid;
-		else
-			rel->rd_node.tblNode = MyDatabaseId;
-
 		RelationInitLockInfo(rel);
+		RelationInitPhysicalAddr(rel);
 	}
 
 	/*
diff --git a/src/backend/utils/init/globals.c b/src/backend/utils/init/globals.c
index 3a5d33724a4..4f0f8b67c24 100644
--- a/src/backend/utils/init/globals.c
+++ b/src/backend/utils/init/globals.c
@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/utils/init/globals.c,v 1.90 2004/05/30 17:58:12 tgl Exp $
+ *	  $PostgreSQL: pgsql/src/backend/utils/init/globals.c,v 1.91 2004/06/18 06:13:54 tgl Exp $
  *
  * NOTES
  *	  Globals used all over the place should be declared here and not
@@ -58,6 +58,8 @@ BackendId	MyBackendId = InvalidBackendId;
 char	   *DatabasePath = NULL;
 Oid			MyDatabaseId = InvalidOid;
 
+Oid			MyDatabaseTableSpace = InvalidOid;
+
 pid_t		PostmasterPid = 0;
 
 /*
diff --git a/src/backend/utils/init/miscinit.c b/src/backend/utils/init/miscinit.c
index 6f9ff5aef4d..b0c5ff82e38 100644
--- a/src/backend/utils/init/miscinit.c
+++ b/src/backend/utils/init/miscinit.c
@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/utils/init/miscinit.c,v 1.126 2004/05/30 23:40:38 neilc Exp $
+ *	  $PostgreSQL: pgsql/src/backend/utils/init/miscinit.c,v 1.127 2004/06/18 06:13:54 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -157,7 +157,6 @@ void
 SetDataDir(const char *dir)
 {
 	char	   *new;
-	int			newlen;
 
 	AssertArg(dir);
 
@@ -212,13 +211,7 @@ SetDataDir(const char *dir)
 	 * Strip any trailing slash.  Not strictly necessary, but avoids
 	 * generating funny-looking paths to individual files.
 	 */
-	newlen = strlen(new);
-	if (newlen > 1 && (new[newlen - 1] == '/'
-#ifdef WIN32
-		|| new[newlen - 1] == '\\'
-#endif
-		))
-		new[newlen - 1] = '\0';
+	canonicalize_path(new);
 
 	if (DataDir)
 		free(DataDir);
diff --git a/src/backend/utils/init/postinit.c b/src/backend/utils/init/postinit.c
index e80187d5752..48d28d429f2 100644
--- a/src/backend/utils/init/postinit.c
+++ b/src/backend/utils/init/postinit.c
@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/utils/init/postinit.c,v 1.133 2004/05/29 22:48:21 tgl Exp $
+ *	  $PostgreSQL: pgsql/src/backend/utils/init/postinit.c,v 1.134 2004/06/18 06:13:54 tgl Exp $
  *
  *
  *-------------------------------------------------------------------------
@@ -26,6 +26,7 @@
 #include "catalog/namespace.h"
 #include "catalog/pg_database.h"
 #include "catalog/pg_shadow.h"
+#include "catalog/pg_tablespace.h"
 #include "commands/trigger.h"
 #include "mb/pg_wchar.h"
 #include "miscadmin.h"
@@ -239,12 +240,12 @@ InitPostgres(const char *dbname, const char *username)
 	if (bootstrap)
 	{
 		MyDatabaseId = TemplateDbOid;
-		SetDatabasePath(GetDatabasePath(MyDatabaseId));
+		MyDatabaseTableSpace = DEFAULTTABLESPACE_OID;
+		SetDatabasePath(GetDatabasePath(MyDatabaseId, MyDatabaseTableSpace));
 	}
 	else
 	{
-		char	   *fullpath,
-					datpath[MAXPGPATH];
+		char	   *fullpath;
 
 		/*
 		 * Formerly we validated DataDir here, but now that's done
@@ -252,11 +253,11 @@ InitPostgres(const char *dbname, const char *username)
 		 */
 
 		/*
-		 * Find oid and path of the database we're about to open. Since
-		 * we're not yet up and running we have to use the hackish
+		 * Find oid and tablespace of the database we're about to open.
+		 * Since we're not yet up and running we have to use the hackish
 		 * GetRawDatabaseInfo.
 		 */
-		GetRawDatabaseInfo(dbname, &MyDatabaseId, datpath);
+		GetRawDatabaseInfo(dbname, &MyDatabaseId, &MyDatabaseTableSpace);
 
 		if (!OidIsValid(MyDatabaseId))
 			ereport(FATAL,
@@ -264,7 +265,7 @@ InitPostgres(const char *dbname, const char *username)
 					 errmsg("database \"%s\" does not exist",
 							dbname)));
 
-		fullpath = GetDatabasePath(MyDatabaseId);
+		fullpath = GetDatabasePath(MyDatabaseId, MyDatabaseTableSpace);
 
 		/* Verify the database path */
 
diff --git a/src/backend/utils/misc/database.c b/src/backend/utils/misc/database.c
index 37844a03a94..1eeb5357040 100644
--- a/src/backend/utils/misc/database.c
+++ b/src/backend/utils/misc/database.c
@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/utils/misc/database.c,v 1.60 2004/01/22 20:57:39 tgl Exp $
+ *	  $PostgreSQL: pgsql/src/backend/utils/misc/database.c,v 1.61 2004/06/18 06:13:56 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -21,6 +21,7 @@
 #include "catalog/catname.h"
 #include "catalog/catalog.h"
 #include "catalog/pg_database.h"
+#include "catalog/pg_tablespace.h"
 #include "miscadmin.h"
 #include "utils/syscache.h"
 
@@ -29,12 +30,13 @@ static bool PhonyHeapTupleSatisfiesNow(HeapTupleHeader tuple);
 
 
 /* --------------------------------
- *	GetRawDatabaseInfo() -- Find the OID and path of the database.
+ *	GetRawDatabaseInfo() -- Find the OID and tablespace of the database.
  *
- *		The database's oid forms half of the unique key for the system
- *		caches and lock tables.  We therefore want it initialized before
- *		we open any relations, since opening relations puts things in the
- *		cache.	To get around this problem, this code opens and scans the
+ *		We need both the OID and the default tablespace in order to find
+ *		the database's system catalogs.  Moreover the database's OID forms
+ *		half of the unique key for the system caches and lock tables, so
+ *		we must have it before we can use any of the cache mechanisms.
+ *		To get around these problems, this code opens and scans the
  *		pg_database relation by hand.
  *
  *		This code knows way more than it should about the layout of
@@ -43,19 +45,21 @@ static bool PhonyHeapTupleSatisfiesNow(HeapTupleHeader tuple);
  * --------------------------------
  */
 void
-GetRawDatabaseInfo(const char *name, Oid *db_id, char *path)
+GetRawDatabaseInfo(const char *name, Oid *db_id, Oid *db_tablespace)
 {
 	int			dbfd;
 	int			nbytes;
-	int			pathlen;
 	HeapTupleData tup;
+	Form_pg_database tup_db;
 	Page		pg;
 	char	   *dbfname;
-	Form_pg_database tup_db;
 	RelFileNode rnode;
 
-	rnode.tblNode = 0;
+	/* hard-wired path to pg_database */
+	rnode.spcNode = GLOBALTABLESPACE_OID;
+	rnode.dbNode = 0;
 	rnode.relNode = RelOid_pg_database;
+
 	dbfname = relpath(rnode);
 
 	if ((dbfd = open(dbfname, O_RDONLY | PG_BINARY, 0)) < 0)
@@ -121,7 +125,7 @@ GetRawDatabaseInfo(const char *name, Oid *db_id, char *path)
 			 * committed and dead tuples to be marked with correct states.
 			 *
 			 * XXX wouldn't it be better to let new backends read the
-			 * database OID from a flat file, handled the same way we
+			 * database info from a flat file, handled the same way we
 			 * handle the password relation?
 			 */
 			if (!PhonyHeapTupleSatisfiesNow(tup.t_data))
@@ -134,15 +138,9 @@ GetRawDatabaseInfo(const char *name, Oid *db_id, char *path)
 
 			if (strcmp(name, NameStr(tup_db->datname)) == 0)
 			{
-				/* Found it; extract the OID and the database path. */
+				/* Found it; extract the db's OID and tablespace. */
 				*db_id = HeapTupleGetOid(&tup);
-				pathlen = VARSIZE(&(tup_db->datpath)) - VARHDRSZ;
-				if (pathlen < 0)
-					pathlen = 0;	/* pure paranoia */
-				if (pathlen >= MAXPGPATH)
-					pathlen = MAXPGPATH - 1;	/* more paranoia */
-				strncpy(path, VARDATA(&(tup_db->datpath)), pathlen);
-				path[pathlen] = '\0';
+				*db_tablespace = tup_db->dattablespace;
 				goto done;
 			}
 		}
@@ -150,7 +148,7 @@ GetRawDatabaseInfo(const char *name, Oid *db_id, char *path)
 
 	/* failed to find it... */
 	*db_id = InvalidOid;
-	*path = '\0';
+	*db_tablespace = InvalidOid;
 
 done:
 	close(dbfd);
diff --git a/src/bin/initdb/initdb.c b/src/bin/initdb/initdb.c
index 3ded3d8775b..44f092fbcc1 100644
--- a/src/bin/initdb/initdb.c
+++ b/src/bin/initdb/initdb.c
@@ -39,7 +39,7 @@
  * Portions Copyright (c) 1994, Regents of the University of California
  * Portions taken from FreeBSD.
  *
- * $PostgreSQL: pgsql/src/bin/initdb/initdb.c,v 1.37 2004/06/10 22:26:20 momjian Exp $
+ * $PostgreSQL: pgsql/src/bin/initdb/initdb.c,v 1.38 2004/06/18 06:13:58 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -1785,7 +1785,7 @@ main(int argc, char *argv[])
 	char	   *pgdenv;			/* PGDATA value got from sent to
 								 * environment */
 	char	   *subdirs[] =
-	{"global", "pg_xlog", "pg_clog", "base", "base/1"};
+	{"global", "pg_xlog", "pg_clog", "base", "base/1", "pg_tablespaces"};
 
 	progname = get_progname(argv[0]);
 	set_pglocale_pgservice(argv[0], "initdb");
@@ -2141,7 +2141,7 @@ main(int argc, char *argv[])
 	/* Bootstrap template1 */
 	bootstrap_template1(short_version);
 
-	/* Make the per-database PGVERSION for template1 only after init'ing it */
+	/* Make the per-database PG_VERSION for template1 only after init'ing it */
 	set_short_version(short_version, "base/1");
 
 	/* Create the stuff we don't need to use bootstrap mode for */
diff --git a/src/bin/pg_dump/dumputils.c b/src/bin/pg_dump/dumputils.c
index 1dfe7801ec5..7c6fd84c5c4 100644
--- a/src/bin/pg_dump/dumputils.c
+++ b/src/bin/pg_dump/dumputils.c
@@ -7,7 +7,7 @@
  * Portions Copyright (c) 1996-2003, PostgreSQL Global Development Group
  * Portions Copyright (c) 1994, Regents of the University of California
  *
- * $PostgreSQL: pgsql/src/bin/pg_dump/dumputils.c,v 1.12 2004/03/23 22:06:08 tgl Exp $
+ * $PostgreSQL: pgsql/src/bin/pg_dump/dumputils.c,v 1.13 2004/06/18 06:14:00 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -144,7 +144,7 @@ appendStringLiteral(PQExpBuffer buf, const char *str, bool escapeAll)
 
 /*
  * Convert a string value to a dollar quoted literal and append it to
- * the given buffer. If the dqprefix parameter is not NULL then the 
+ * the given buffer. If the dqprefix parameter is not NULL then the
  * dollar quote delimiter will begin with that (after the opening $).
  *
  * No escaping is done at all on str, in compliance with the rules
@@ -162,7 +162,7 @@ appendStringLiteralDQ(PQExpBuffer buf, const char *str, const char *dqprefix)
 	if (dqprefix)
 		appendPQExpBuffer(delimBuf, dqprefix);
 
-	/* 
+	/*
 	 * Make sure we choose a delimiter which (without the trailing $)
 	 * is not present in the string being quoted. We don't check with the
 	 * trailing $ because a string ending in $foo must not be quoted with
@@ -191,7 +191,7 @@ appendStringLiteralDQ(PQExpBuffer buf, const char *str, const char *dqprefix)
  * otherwise use standard quoting.
  */
 void
-appendStringLiteralDQOpt(PQExpBuffer buf, const char *str, 
+appendStringLiteralDQOpt(PQExpBuffer buf, const char *str,
 						 bool escapeAll, const char *dqprefix)
 {
 	if (strchr(str, '\'') == NULL && strchr(str, '\\') == NULL)
@@ -586,6 +586,8 @@ parseAclItem(const char *item, const char *type, const char *name,
 		CONVERT_PRIV('C', "CREATE");
 		CONVERT_PRIV('T', "TEMPORARY");
 	}
+	else if (strcmp(type, "TABLESPACE") == 0)
+		CONVERT_PRIV('C', "CREATE");
 	else
 		abort();
 
@@ -624,7 +626,7 @@ copyAclUserName(PQExpBuffer output, char *input)
 			appendPQExpBufferChar(output, *input++);
 		else
 		{
-			/* Otherwise, it's a quoted username */ 
+			/* Otherwise, it's a quoted username */
 			input++;
 			/* Loop until we come across an unescaped quote */
 			while (!(*input == '"' && *(input + 1) != '"'))
diff --git a/src/bin/pg_dump/pg_dump.c b/src/bin/pg_dump/pg_dump.c
index 292e70a6db8..c5320177fa3 100644
--- a/src/bin/pg_dump/pg_dump.c
+++ b/src/bin/pg_dump/pg_dump.c
@@ -12,7 +12,7 @@
  *	by PostgreSQL
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/bin/pg_dump/pg_dump.c,v 1.374 2004/06/07 20:35:57 momjian Exp $
+ *	  $PostgreSQL: pgsql/src/bin/pg_dump/pg_dump.c,v 1.375 2004/06/18 06:14:00 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -1164,13 +1164,13 @@ dumpDatabase(Archive *AH)
 				i_oid,
 				i_dba,
 				i_encoding,
-				i_datpath;
+				i_tablespace;
 	CatalogId	dbCatId;
 	DumpId		dbDumpId;
 	const char *datname,
 			   *dba,
 			   *encoding,
-			   *datpath;
+			   *tablespace;
 
 	datname = PQdb(g_conn);
 
@@ -1181,31 +1181,34 @@ dumpDatabase(Archive *AH)
 	selectSourceSchema("pg_catalog");
 
 	/* Get the database owner and parameters from pg_database */
-	if (g_fout->remoteVersion >= 70100)
+	if (g_fout->remoteVersion >= 70500)
 	{
 		appendPQExpBuffer(dbQry, "SELECT tableoid, oid, "
 						  "(SELECT usename FROM pg_user WHERE usesysid = datdba) as dba, "
 						  "pg_encoding_to_char(encoding) as encoding, "
-						  "datpath "
+						  "(SELECT spcname FROM pg_tablespace t WHERE t.oid = dattablespace) as tablespace "
+						  "FROM pg_database "
+						  "WHERE datname = ");
+		appendStringLiteral(dbQry, datname, true);
+	}
+	else if (g_fout->remoteVersion >= 70100)
+	{
+		appendPQExpBuffer(dbQry, "SELECT tableoid, oid, "
+						  "(SELECT usename FROM pg_user WHERE usesysid = datdba) as dba, "
+						  "pg_encoding_to_char(encoding) as encoding, "
+						  "NULL as tablespace "
 						  "FROM pg_database "
 						  "WHERE datname = ");
 		appendStringLiteral(dbQry, datname, true);
 	}
 	else
 	{
-		/*
-		 * In 7.0, datpath is either the same as datname, or the user-given
-		 * location with "/" and the datname appended.  We must strip this
-		 * junk off to produce a correct LOCATION value.
-		 */
 		appendPQExpBuffer(dbQry, "SELECT "
 						  "(SELECT oid FROM pg_class WHERE relname = 'pg_database') AS tableoid, "
 						  "oid, "
 						  "(SELECT usename FROM pg_user WHERE usesysid = datdba) as dba, "
 						  "pg_encoding_to_char(encoding) as encoding, "
-						  "CASE WHEN length(datpath) > length(datname) THEN "
-						  "substr(datpath,1,length(datpath)-length(datname)-1) "
-						  "ELSE '' END as datpath "
+						  "NULL as tablespace "
 						  "FROM pg_database "
 						  "WHERE datname = ");
 		appendStringLiteral(dbQry, datname, true);
@@ -1234,26 +1237,25 @@ dumpDatabase(Archive *AH)
 	i_oid = PQfnumber(res, "oid");
 	i_dba = PQfnumber(res, "dba");
 	i_encoding = PQfnumber(res, "encoding");
-	i_datpath = PQfnumber(res, "datpath");
+	i_tablespace = PQfnumber(res, "tablespace");
 
 	dbCatId.tableoid = atooid(PQgetvalue(res, 0, i_tableoid));
 	dbCatId.oid = atooid(PQgetvalue(res, 0, i_oid));
 	dba = PQgetvalue(res, 0, i_dba);
 	encoding = PQgetvalue(res, 0, i_encoding);
-	datpath = PQgetvalue(res, 0, i_datpath);
+	tablespace = PQgetvalue(res, 0, i_tablespace);
 
 	appendPQExpBuffer(creaQry, "CREATE DATABASE %s WITH TEMPLATE = template0",
 					  fmtId(datname));
-	if (strlen(datpath) > 0)
-	{
-		appendPQExpBuffer(creaQry, " LOCATION = ");
-		appendStringLiteral(creaQry, datpath, true);
-	}
 	if (strlen(encoding) > 0)
 	{
 		appendPQExpBuffer(creaQry, " ENCODING = ");
 		appendStringLiteral(creaQry, encoding, true);
 	}
+	if (strlen(tablespace) > 0 && strcmp(tablespace, "default") != 0)
+	{
+		appendPQExpBuffer(creaQry, " TABLESPACE = %s", fmtId(tablespace));
+	}
 	appendPQExpBuffer(creaQry, ";\n");
 
 	appendPQExpBuffer(delQry, "DROP DATABASE %s;\n",
@@ -1303,7 +1305,7 @@ dumpTimestamp(Archive *AH, char *msg)
 	if (strftime(buf, 256, "%Y-%m-%d %H:%M:%S %Z", localtime(&now)) != 0)
  	{
 		PQExpBuffer qry = createPQExpBuffer();
-	
+
 		appendPQExpBuffer(qry, "-- ");
 		appendPQExpBuffer(qry, msg);
 		appendPQExpBuffer(qry, " ");
@@ -1471,6 +1473,7 @@ getNamespaces(int *numNamespaces)
 	int			i_oid;
 	int			i_nspname;
 	int			i_usename;
+	int			i_nsptablespace;
 	int			i_nspacl;
 
 	/*
@@ -1488,6 +1491,7 @@ getNamespaces(int *numNamespaces)
 		nsinfo[0].dobj.name = strdup("");
 		nsinfo[0].usename = strdup("");
 		nsinfo[0].nspacl = strdup("");
+		nsinfo[0].nsptablespace = strdup("");
 
 		selectDumpableNamespace(&nsinfo[0]);
 
@@ -1498,6 +1502,7 @@ getNamespaces(int *numNamespaces)
 		nsinfo[1].dobj.name = strdup("pg_catalog");
 		nsinfo[1].usename = strdup("");
 		nsinfo[1].nspacl = strdup("");
+		nsinfo[0].nsptablespace = strdup("");
 
 		selectDumpableNamespace(&nsinfo[1]);
 
@@ -1516,10 +1521,21 @@ getNamespaces(int *numNamespaces)
 	 * we fetch all namespaces including system ones, so that every object
 	 * we read in can be linked to a containing namespace.
 	 */
-	appendPQExpBuffer(query, "SELECT tableoid, oid, nspname, "
-	"(select usename from pg_user where nspowner = usesysid) as usename, "
-					  "nspacl "
-					  "FROM pg_namespace");
+	if (g_fout->remoteVersion >= 70500)
+	{
+		appendPQExpBuffer(query, "SELECT tableoid, oid, nspname, "
+		"(select usename from pg_user where nspowner = usesysid) as usename, "
+						  "nspacl, "
+						 "(SELECT spcname FROM pg_tablespace t WHERE t.oid = nsptablespace) AS nsptablespace "
+						  "FROM pg_namespace");
+	}
+	else
+	{
+		appendPQExpBuffer(query, "SELECT tableoid, oid, nspname, "
+		"(select usename from pg_user where nspowner = usesysid) as usename, "
+						  "nspacl, NULL AS nsptablespace "
+						  "FROM pg_namespace");
+	}
 
 	res = PQexec(g_conn, query->data);
 	check_sql_result(res, g_conn, query->data, PGRES_TUPLES_OK);
@@ -1533,6 +1549,7 @@ getNamespaces(int *numNamespaces)
 	i_nspname = PQfnumber(res, "nspname");
 	i_usename = PQfnumber(res, "usename");
 	i_nspacl = PQfnumber(res, "nspacl");
+	i_nsptablespace = PQfnumber(res, "nsptablespace");
 
 	for (i = 0; i < ntups; i++)
 	{
@@ -1543,6 +1560,7 @@ getNamespaces(int *numNamespaces)
 		nsinfo[i].dobj.name = strdup(PQgetvalue(res, i, i_nspname));
 		nsinfo[i].usename = strdup(PQgetvalue(res, i, i_usename));
 		nsinfo[i].nspacl = strdup(PQgetvalue(res, i, i_nspacl));
+		nsinfo[i].nsptablespace = strdup(PQgetvalue(res, i, i_nsptablespace));
 
 		/* Decide whether to dump this namespace */
 		selectDumpableNamespace(&nsinfo[i]);
@@ -2329,6 +2347,7 @@ getTables(int *numTables)
 	int			i_relhasoids;
 	int			i_owning_tab;
 	int			i_owning_col;
+	int			i_reltablespace;
 
 	/* Make sure we are in proper schema */
 	selectSourceSchema("pg_catalog");
@@ -2349,7 +2368,7 @@ getTables(int *numTables)
 	 * columns, etc.
 	 */
 
-	if (g_fout->remoteVersion >= 70300)
+	if (g_fout->remoteVersion >= 70500)
 	{
 		/*
 		 * Left join to pick up dependency info linking sequences to their
@@ -2362,7 +2381,34 @@ getTables(int *numTables)
 						  "relchecks, reltriggers, "
 						  "relhasindex, relhasrules, relhasoids, "
 						  "d.refobjid as owning_tab, "
-						  "d.refobjsubid as owning_col "
+						  "d.refobjsubid as owning_col, "
+						  "(SELECT spcname FROM pg_tablespace t WHERE t.oid = c.reltablespace) AS reltablespace "
+						  "from pg_class c "
+						  "left join pg_depend d on "
+						  "(c.relkind = '%c' and "
+						"d.classid = c.tableoid and d.objid = c.oid and "
+						  "d.objsubid = 0 and "
+						"d.refclassid = c.tableoid and d.deptype = 'i') "
+						  "where relkind in ('%c', '%c', '%c') "
+						  "order by c.oid",
+						  RELKIND_SEQUENCE,
+					   RELKIND_RELATION, RELKIND_SEQUENCE, RELKIND_VIEW);
+	}
+	else if (g_fout->remoteVersion >= 70300)
+	{
+		/*
+		 * Left join to pick up dependency info linking sequences to their
+		 * serial column, if any
+		 */
+		appendPQExpBuffer(query,
+						  "SELECT c.tableoid, c.oid, relname, "
+						  "relacl, relkind, relnamespace, "
+						  "(select usename from pg_user where relowner = usesysid) as usename, "
+						  "relchecks, reltriggers, "
+						  "relhasindex, relhasrules, relhasoids, "
+						  "d.refobjid as owning_tab, "
+						  "d.refobjsubid as owning_col, "
+						  "NULL as reltablespace "
 						  "from pg_class c "
 						  "left join pg_depend d on "
 						  "(c.relkind = '%c' and "
@@ -2383,7 +2429,8 @@ getTables(int *numTables)
 						  "relchecks, reltriggers, "
 						  "relhasindex, relhasrules, relhasoids, "
 						  "NULL::oid as owning_tab, "
-						  "NULL::int4 as owning_col "
+						  "NULL::int4 as owning_col, "
+						  "NULL as reltablespace "
 						  "from pg_class "
 						  "where relkind in ('%c', '%c', '%c') "
 						  "order by oid",
@@ -2400,7 +2447,8 @@ getTables(int *numTables)
 						  "relhasindex, relhasrules, "
 						  "'t'::bool as relhasoids, "
 						  "NULL::oid as owning_tab, "
-						  "NULL::int4 as owning_col "
+						  "NULL::int4 as owning_col, "
+						  "NULL as reltablespace "
 						  "from pg_class "
 						  "where relkind in ('%c', '%c', '%c') "
 						  "order by oid",
@@ -2427,7 +2475,8 @@ getTables(int *numTables)
 						  "relhasindex, relhasrules, "
 						  "'t'::bool as relhasoids, "
 						  "NULL::oid as owning_tab, "
-						  "NULL::int4 as owning_col "
+						  "NULL::int4 as owning_col, "
+						  "NULL as reltablespace "
 						  "from pg_class c "
 						  "where relkind in ('%c', '%c') "
 						  "order by oid",
@@ -2467,6 +2516,7 @@ getTables(int *numTables)
 	i_relhasoids = PQfnumber(res, "relhasoids");
 	i_owning_tab = PQfnumber(res, "owning_tab");
 	i_owning_col = PQfnumber(res, "owning_col");
+	i_reltablespace = PQfnumber(res, "reltablespace");
 
 	for (i = 0; i < ntups; i++)
 	{
@@ -2495,6 +2545,7 @@ getTables(int *numTables)
 			tblinfo[i].owning_tab = atooid(PQgetvalue(res, i, i_owning_tab));
 			tblinfo[i].owning_col = atoi(PQgetvalue(res, i, i_owning_col));
 		}
+		tblinfo[i].reltablespace = strdup(PQgetvalue(res, i, i_reltablespace));
 
 		/* other fields were zeroed above */
 
@@ -2768,6 +2819,7 @@ getIndexes(TableInfo tblinfo[], int numTables)
 			indxinfo[j].indextable = tbinfo;
 			indxinfo[j].indexdef = strdup(PQgetvalue(res, j, i_indexdef));
 			indxinfo[j].indnkeys = atoi(PQgetvalue(res, j, i_indnkeys));
+
 			/*
 			 * In pre-7.4 releases, indkeys may contain more entries than
 			 * indnkeys says (since indnkeys will be 1 for a functional
@@ -2805,7 +2857,7 @@ getIndexes(TableInfo tblinfo[], int numTables)
 				constrinfo[j].conindex = indxinfo[j].dobj.dumpId;
 				constrinfo[j].coninherited = false;
 				constrinfo[j].separate = true;
-				
+
 				indxinfo[j].indexconstraint = constrinfo[j].dobj.dumpId;
 
 				/* If pre-7.3 DB, better make sure table comes first */
@@ -4341,9 +4393,16 @@ dumpNamespace(Archive *fout, NamespaceInfo *nspinfo)
 	{
 		appendPQExpBuffer(delq, "DROP SCHEMA %s;\n", qnspname);
 
-		appendPQExpBuffer(q, "CREATE SCHEMA %s AUTHORIZATION %s;\n",
+		appendPQExpBuffer(q, "CREATE SCHEMA %s AUTHORIZATION %s",
 						  qnspname, fmtId(nspinfo->usename));
 
+		/* Add tablespace qualifier, if not default */
+		if (strlen(nspinfo->nsptablespace) != 0)
+			appendPQExpBuffer(q, " TABLESPACE %s",
+							  fmtId(nspinfo->nsptablespace));
+
+		appendPQExpBuffer(q, ";\n");
+
 		ArchiveEntry(fout, nspinfo->dobj.catId, nspinfo->dobj.dumpId,
 					 nspinfo->dobj.name,
 					 NULL, "",
@@ -5118,7 +5177,7 @@ dumpFunc(Archive *fout, FuncInfo *finfo)
 		if (strcmp(prosrc, "-") != 0)
 		{
 			appendPQExpBuffer(asPart, ", ");
-			/* 
+			/*
 			 * where we have bin, use dollar quoting if allowed and src
 			 * contains quote or backslash; else use regular quoting.
 			 */
@@ -5281,7 +5340,7 @@ dumpCast(Archive *fout, CastInfo *cast)
 	 * Skip cast if function isn't from pg_ and that namespace is
 	 * not dumped.
 	 */
-	if (funcInfo && 
+	if (funcInfo &&
 		strncmp(funcInfo->dobj.namespace->dobj.name, "pg_", 3) != 0 &&
 		!funcInfo->dobj.namespace->dump)
 		return;
@@ -6001,14 +6060,14 @@ dumpConversion(Archive *fout, ConvInfo *convinfo)
 					  fmtId(convinfo->dobj.name));
 
 	appendPQExpBuffer(q, "CREATE %sCONVERSION %s FOR ",
-					(condefault) ? "DEFAULT " : "",	
+					(condefault) ? "DEFAULT " : "",
 					fmtId(convinfo->dobj.name));
 	appendStringLiteral(q, conforencoding, true);
 	appendPQExpBuffer(q, " TO ");
 	appendStringLiteral(q, contoencoding, true);
 	/* regproc is automatically quoted in 7.3 and above */
 	appendPQExpBuffer(q, " FROM %s;\n", conproc);
-	
+
 	ArchiveEntry(fout, convinfo->dobj.catId, convinfo->dobj.dumpId,
 				 convinfo->dobj.name,
 				 convinfo->dobj.namespace->dobj.name, convinfo->usename,
@@ -6562,6 +6621,15 @@ dumpTableSchema(Archive *fout, TableInfo *tbinfo)
 			appendPQExpBuffer(q, ")");
 		}
 
+		/* Output tablespace clause if necessary */
+		if (strlen(tbinfo->reltablespace) != 0 &&
+			strcmp(tbinfo->reltablespace,
+				   tbinfo->dobj.namespace->nsptablespace) != 0)
+		{
+			appendPQExpBuffer(q, " TABLESPACE %s",
+							  fmtId(tbinfo->reltablespace));
+		}
+
 		appendPQExpBuffer(q, ";\n");
 
 		/* Loop dumping statistics and storage statements */
@@ -7227,9 +7295,20 @@ dumpSequence(Archive *fout, TableInfo *tbinfo)
 			appendPQExpBuffer(query, "    NO MINVALUE\n");
 
 		appendPQExpBuffer(query,
-						  "    CACHE %s%s;\n",
+						  "    CACHE %s%s",
 						  cache, (cycled ? "\n    CYCLE" : ""));
 
+		/* Output tablespace clause if necessary */
+		if (strlen(tbinfo->reltablespace) != 0 &&
+			strcmp(tbinfo->reltablespace,
+				   tbinfo->dobj.namespace->nsptablespace) != 0)
+		{
+			appendPQExpBuffer(query, " TABLESPACE %s",
+							  fmtId(tbinfo->reltablespace));
+		}
+
+		appendPQExpBuffer(query, ";\n");
+
 		ArchiveEntry(fout, tbinfo->dobj.catId, tbinfo->dobj.dumpId,
 					 tbinfo->dobj.name,
 					 tbinfo->dobj.namespace->dobj.name, tbinfo->usename,
diff --git a/src/bin/pg_dump/pg_dump.h b/src/bin/pg_dump/pg_dump.h
index 0f335544d55..55a25212257 100644
--- a/src/bin/pg_dump/pg_dump.h
+++ b/src/bin/pg_dump/pg_dump.h
@@ -6,7 +6,7 @@
  * Portions Copyright (c) 1996-2003, PostgreSQL Global Development Group
  * Portions Copyright (c) 1994, Regents of the University of California
  *
- * $PostgreSQL: pgsql/src/bin/pg_dump/pg_dump.h,v 1.108 2004/03/03 21:28:55 tgl Exp $
+ * $PostgreSQL: pgsql/src/bin/pg_dump/pg_dump.h,v 1.109 2004/06/18 06:14:00 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -98,6 +98,7 @@ typedef struct _namespaceInfo
 	DumpableObject dobj;
 	char	   *usename;		/* name of owner, or empty string */
 	char	   *nspacl;
+	char	   *nsptablespace;	/* default tablespace */
 	bool		dump;			/* true if need to dump definition */
 } NamespaceInfo;
 
@@ -168,6 +169,7 @@ typedef struct _tableInfo
 	char	   *usename;		/* name of owner, or empty string */
 	char	   *relacl;
 	char		relkind;
+	char	   *reltablespace;	/* relation tablespace */
 	bool		hasindex;		/* does it have any indexes? */
 	bool		hasrules;		/* does it have any rules? */
 	bool		hasoids;		/* does it have OIDs? */
diff --git a/src/bin/pg_dump/pg_dumpall.c b/src/bin/pg_dump/pg_dumpall.c
index 11678df3303..3006505172d 100644
--- a/src/bin/pg_dump/pg_dumpall.c
+++ b/src/bin/pg_dump/pg_dumpall.c
@@ -6,7 +6,7 @@
  * Portions Copyright (c) 1994, Regents of the University of California
  *
  *
- * $PostgreSQL: pgsql/src/bin/pg_dump/pg_dumpall.c,v 1.41 2004/06/10 16:35:17 momjian Exp $
+ * $PostgreSQL: pgsql/src/bin/pg_dump/pg_dumpall.c,v 1.42 2004/06/18 06:14:00 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -46,6 +46,7 @@ static void help(void);
 
 static void dumpUsers(PGconn *conn);
 static void dumpGroups(PGconn *conn);
+static void dumpTablespaces(PGconn *conn);
 static void dumpCreateDB(PGconn *conn);
 static void dumpDatabaseConfig(PGconn *conn, const char *dbname);
 static void dumpUserConfig(PGconn *conn, const char *username);
@@ -231,6 +232,8 @@ main(int argc, char *argv[])
 	{
 		dumpUsers(conn);
 		dumpGroups(conn);
+		if (server_version >= 70500)
+			dumpTablespaces(conn);
 	}
 
 	if (!globals_only)
@@ -411,7 +414,68 @@ dumpGroups(PGconn *conn)
 	printf("\n\n");
 }
 
+/*
+ * Dump tablespaces.
+ */
+static void
+dumpTablespaces(PGconn *conn)
+{
+	PGresult   *res;
+	int			i;
+
+	printf("--\n-- Tablespaces\n--\n\n");
+
+	/*
+	 * Get all tablespaces except for the system default and global
+	 * tablespaces
+	 */
+	res = executeQuery(conn, "SELECT spcname, "
+					   "pg_catalog.pg_get_userbyid(spcowner) AS spcowner, "
+					   "spclocation, spcacl "
+					   "FROM pg_catalog.pg_tablespace "
+					   "WHERE spcname NOT IN ('default', 'global')");
+
+	for (i = 0; i < PQntuples(res); i++)
+	{
+		PQExpBuffer buf = createPQExpBuffer();
+		char	*spcname = PQgetvalue(res, i, 0);
+		char	*spcowner = PQgetvalue(res, i, 1);
+		char	*spclocation = PQgetvalue(res, i, 2);
+		char	*spcacl = PQgetvalue(res, i, 3);
+		char	*fspcname;
+
+		/* needed for buildACLCommands() */
+		fspcname = strdup(fmtId(spcname));
+
+		if (output_clean)
+			appendPQExpBuffer(buf, "DROP TABLESPACE %s;\n", fspcname);
+
+		appendPQExpBuffer(buf, "CREATE TABLESPACE %s", fspcname);
+		appendPQExpBuffer(buf, " OWNER %s", fmtId(spcowner));
+
+		appendPQExpBuffer(buf, " LOCATION ");
+		appendStringLiteral(buf, spclocation, true);
+		appendPQExpBuffer(buf, ";\n");
+
+		if (!skip_acls &&
+			!buildACLCommands(fspcname, "TABLESPACE", spcacl, spcowner,
+							  server_version, buf))
+		{
+			fprintf(stderr, _("%s: could not parse ACL list (%s) for tablespace \"%s\"\n"),
+					progname, spcacl, fspcname);
+			PQfinish(conn);
+			exit(1);
+		}
+
+		printf("%s", buf->data);
 
+		free(fspcname);
+		destroyPQExpBuffer(buf);
+	}
+
+	PQclear(res);
+	printf("\n\n");
+}
 
 /*
  * Dump commands to create each database.
@@ -432,12 +496,22 @@ dumpCreateDB(PGconn *conn)
 
 	printf("--\n-- Database creation\n--\n\n");
 
-	if (server_version >= 70300)
+	if (server_version >= 70500)
 		res = executeQuery(conn,
 						   "SELECT datname, "
 						   "coalesce(usename, (select usename from pg_shadow where usesysid=(select datdba from pg_database where datname='template0'))), "
 						   "pg_encoding_to_char(d.encoding), "
-						   "datistemplate, datpath, datacl "
+						   "datistemplate, datacl, "
+						   "(SELECT spcname FROM pg_tablespace t WHERE t.oid = d.dattablespace) AS dattablespace "
+		"FROM pg_database d LEFT JOIN pg_shadow u ON (datdba = usesysid) "
+						   "WHERE datallowconn ORDER BY 1");
+	else if (server_version >= 70300)
+		res = executeQuery(conn,
+						   "SELECT datname, "
+						   "coalesce(usename, (select usename from pg_shadow where usesysid=(select datdba from pg_database where datname='template0'))), "
+						   "pg_encoding_to_char(d.encoding), "
+						   "datistemplate, datacl, "
+						   "'default' AS dattablespace "
 		"FROM pg_database d LEFT JOIN pg_shadow u ON (datdba = usesysid) "
 						   "WHERE datallowconn ORDER BY 1");
 	else if (server_version >= 70100)
@@ -447,16 +521,13 @@ dumpCreateDB(PGconn *conn)
 				"(select usename from pg_shadow where usesysid=datdba), "
 						   "(select usename from pg_shadow where usesysid=(select datdba from pg_database where datname='template0'))), "
 						   "pg_encoding_to_char(d.encoding), "
-						   "datistemplate, datpath, '' as datacl "
+						   "datistemplate, '' as datacl, "
+						   "'default' AS dattablespace "
 						   "FROM pg_database d "
 						   "WHERE datallowconn ORDER BY 1");
 	else
 	{
 		/*
-		 * In 7.0, datpath is either the same as datname, or the user-given
-		 * location with "/" and the datname appended.  We must strip this
-		 * junk off to produce a correct LOCATION value.
-		 *
 		 * Note: 7.0 fails to cope with sub-select in COALESCE, so just
 		 * deal with getting a NULL by not printing any OWNER clause.
 		 */
@@ -465,10 +536,8 @@ dumpCreateDB(PGconn *conn)
 				"(select usename from pg_shadow where usesysid=datdba), "
 						   "pg_encoding_to_char(d.encoding), "
 						   "'f' as datistemplate, "
-						   "CASE WHEN length(datpath) > length(datname) THEN "
-						   "substr(datpath,1,length(datpath)-length(datname)-1) "
-						   "ELSE '' END as datpath, "
-						   "'' as datacl "
+						   "'' as datacl, "
+						   "'default' AS dattablespace "
 						   "FROM pg_database d "
 						   "ORDER BY 1");
 	}
@@ -480,8 +549,8 @@ dumpCreateDB(PGconn *conn)
 		char	   *dbowner = PQgetvalue(res, i, 1);
 		char	   *dbencoding = PQgetvalue(res, i, 2);
 		char	   *dbistemplate = PQgetvalue(res, i, 3);
-		char	   *dbpath = PQgetvalue(res, i, 4);
-		char	   *dbacl = PQgetvalue(res, i, 5);
+		char	   *dbacl = PQgetvalue(res, i, 4);
+		char	   *dbtablespace = PQgetvalue(res, i, 5);
 		char	   *fdbname;
 
 		if (strcmp(dbname, "template1") == 0)
@@ -496,20 +565,21 @@ dumpCreateDB(PGconn *conn)
 			appendPQExpBuffer(buf, "DROP DATABASE %s;\n", fdbname);
 
 		appendPQExpBuffer(buf, "CREATE DATABASE %s", fdbname);
+
+		appendPQExpBuffer(buf, " WITH TEMPLATE = template0");
+
 		if (strlen(dbowner) != 0)
-			appendPQExpBuffer(buf, " WITH OWNER = %s",
+			appendPQExpBuffer(buf, " OWNER = %s",
 							  fmtId(dbowner));
-		appendPQExpBuffer(buf, " TEMPLATE = template0");
-
-		if (strlen(dbpath) != 0)
-		{
-			appendPQExpBuffer(buf, " LOCATION = ");
-			appendStringLiteral(buf, dbpath, true);
-		}
 
 		appendPQExpBuffer(buf, " ENCODING = ");
 		appendStringLiteral(buf, dbencoding, true);
 
+		/* Output tablespace if it isn't default */
+		if (strcmp(dbtablespace, "default") != 0)
+			appendPQExpBuffer(buf, " TABLESPACE = %s",
+							  fmtId(dbtablespace));
+
 		appendPQExpBuffer(buf, ";\n");
 
 		if (strcmp(dbistemplate, "t") == 0)
diff --git a/src/bin/psql/command.c b/src/bin/psql/command.c
index d880afbd276..af8c2254484 100644
--- a/src/bin/psql/command.c
+++ b/src/bin/psql/command.c
@@ -3,7 +3,7 @@
  *
  * Copyright (c) 2000-2003, PostgreSQL Global Development Group
  *
- * $PostgreSQL: pgsql/src/bin/psql/command.c,v 1.116 2004/05/07 00:24:58 tgl Exp $
+ * $PostgreSQL: pgsql/src/bin/psql/command.c,v 1.117 2004/06/18 06:14:04 tgl Exp $
  */
 #include "postgres_fe.h"
 #include "command.h"
@@ -301,6 +301,9 @@ exec_command(const char *cmd,
 			case 'a':
 				success = describeAggregates(pattern, show_verbose);
 				break;
+			case 'b':
+				success = describeTablespaces(pattern);
+				break;
 			case 'c':
 				success = listConversions(pattern);
 				break;
diff --git a/src/bin/psql/describe.c b/src/bin/psql/describe.c
index e06c2001581..0f75d97d23d 100644
--- a/src/bin/psql/describe.c
+++ b/src/bin/psql/describe.c
@@ -3,7 +3,7 @@
  *
  * Copyright (c) 2000-2003, PostgreSQL Global Development Group
  *
- * $PostgreSQL: pgsql/src/bin/psql/describe.c,v 1.98 2004/05/07 00:24:58 tgl Exp $
+ * $PostgreSQL: pgsql/src/bin/psql/describe.c,v 1.99 2004/06/18 06:14:04 tgl Exp $
  */
 #include "postgres_fe.h"
 #include "describe.h"
@@ -99,6 +99,45 @@ describeAggregates(const char *pattern, bool verbose)
 	return true;
 }
 
+/* \db
+ * Takes an optional regexp to select particular tablespaces
+ */
+bool
+describeTablespaces(const char *pattern)
+{
+	PQExpBufferData buf;
+	PGresult   *res;
+	printQueryOpt myopt = pset.popt;
+
+	initPQExpBuffer(&buf);
+
+	printfPQExpBuffer(&buf,
+					  "SELECT spcname AS \"%s\",\n"
+					  "  pg_catalog.pg_get_userbyid(spcowner) AS \"%s\",\n"
+					  "  spclocation AS \"%s\"\n"
+					  "FROM pg_catalog.pg_tablespace\n",
+					  _("Name"), _("Owner"), _("Location"));
+
+	processNamePattern(&buf, pattern, false, false,
+					   NULL, "spcname", NULL,
+					   NULL);
+
+	appendPQExpBuffer(&buf, "ORDER BY 1;");
+
+	res = PSQLexec(buf.data, false);
+	termPQExpBuffer(&buf);
+	if (!res)
+		return false;
+
+	myopt.nullPrint = NULL;
+	myopt.title = _("List of tablespaces");
+
+	printQuery(res, &myopt, pset.queryFout);
+
+	PQclear(res);
+	return true;
+}
+
 
 /* \df
  * Takes an optional regexp to select particular functions
@@ -351,7 +390,7 @@ permissionsList(const char *pattern)
 	printfPQExpBuffer(&buf,
 					  "SELECT n.nspname as \"%s\",\n"
 					  "  c.relname as \"%s\",\n"
-					  "  CASE c.relkind WHEN 'r' THEN '%s' WHEN 'v' THEN '%s' WHEN 'S' THEN '%s' END as \"%s\",\n" 
+					  "  CASE c.relkind WHEN 'r' THEN '%s' WHEN 'v' THEN '%s' WHEN 'S' THEN '%s' END as \"%s\",\n"
 					  "  c.relacl as \"%s\"\n"
 					  "FROM pg_catalog.pg_class c\n"
 	"     LEFT JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace\n"
diff --git a/src/bin/psql/describe.h b/src/bin/psql/describe.h
index acc841a02b2..195bf0e5578 100644
--- a/src/bin/psql/describe.h
+++ b/src/bin/psql/describe.h
@@ -3,7 +3,7 @@
  *
  * Copyright (c) 2000-2003, PostgreSQL Global Development Group
  *
- * $PostgreSQL: pgsql/src/bin/psql/describe.h,v 1.23 2003/12/01 22:21:54 momjian Exp $
+ * $PostgreSQL: pgsql/src/bin/psql/describe.h,v 1.24 2004/06/18 06:14:04 tgl Exp $
  */
 #ifndef DESCRIBE_H
 #define DESCRIBE_H
@@ -13,6 +13,9 @@
 /* \da */
 bool		describeAggregates(const char *pattern, bool verbose);
 
+/* \db */
+bool		describeTablespaces(const char *pattern);
+
 /* \df */
 bool		describeFunctions(const char *pattern, bool verbose);
 
diff --git a/src/bin/psql/help.c b/src/bin/psql/help.c
index f7e9022e425..ed9574a2297 100644
--- a/src/bin/psql/help.c
+++ b/src/bin/psql/help.c
@@ -3,7 +3,7 @@
  *
  * Copyright (c) 2000-2003, PostgreSQL Global Development Group
  *
- * $PostgreSQL: pgsql/src/bin/psql/help.c,v 1.87 2004/05/07 00:24:58 tgl Exp $
+ * $PostgreSQL: pgsql/src/bin/psql/help.c,v 1.88 2004/06/18 06:14:04 tgl Exp $
  */
 #include "postgres_fe.h"
 #include "common.h"
@@ -211,6 +211,7 @@ slashUsage(unsigned short int pager)
 	fprintf(output, _("  \\d{t|i|s|v|S} [PATTERN] (add \"+\" for more detail)\n"
 					  "                 list tables/indexes/sequences/views/system tables\n"));
 	fprintf(output, _("  \\da [PATTERN]  list aggregate functions\n"));
+	fprintf(output, _("  \\db [PATTERN]  list tablespaces\n"));
 	fprintf(output, _("  \\dc [PATTERN]  list conversions\n"));
 	fprintf(output, _("  \\dC            list casts\n"));
 	fprintf(output, _("  \\dd [PATTERN]  show comment for object\n"));
@@ -308,7 +309,7 @@ helpSQL(const char *topic, unsigned short int pager)
 		size_t		len;
 		int			nl_count = 0;
 		char		*ch;
-		
+
 		/* don't care about trailing spaces */
 		len = strlen(topic);
 		while (topic[len - 1] == ' ')
diff --git a/src/include/catalog/catalog.h b/src/include/catalog/catalog.h
index 293e11ef025..15cb6e7ca07 100644
--- a/src/include/catalog/catalog.h
+++ b/src/include/catalog/catalog.h
@@ -7,7 +7,7 @@
  * Portions Copyright (c) 1996-2003, PostgreSQL Global Development Group
  * Portions Copyright (c) 1994, Regents of the University of California
  *
- * $PostgreSQL: pgsql/src/include/catalog/catalog.h,v 1.27 2003/11/29 22:40:58 pgsql Exp $
+ * $PostgreSQL: pgsql/src/include/catalog/catalog.h,v 1.28 2004/06/18 06:14:05 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -18,7 +18,7 @@
 
 
 extern char *relpath(RelFileNode rnode);
-extern char *GetDatabasePath(Oid tblNode);
+extern char *GetDatabasePath(Oid dbNode, Oid spcNode);
 
 extern bool IsSystemRelation(Relation relation);
 extern bool IsToastRelation(Relation relation);
diff --git a/src/include/catalog/catname.h b/src/include/catalog/catname.h
index 7b3b99701a9..240f4947082 100644
--- a/src/include/catalog/catname.h
+++ b/src/include/catalog/catname.h
@@ -7,7 +7,7 @@
  * Portions Copyright (c) 1996-2003, PostgreSQL Global Development Group
  * Portions Copyright (c) 1994, Regents of the University of California
  *
- * $PostgreSQL: pgsql/src/include/catalog/catname.h,v 1.31 2003/11/29 22:40:58 pgsql Exp $
+ * $PostgreSQL: pgsql/src/include/catalog/catname.h,v 1.32 2004/06/18 06:14:05 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -40,6 +40,7 @@
 #define  RewriteRelationName "pg_rewrite"
 #define  ShadowRelationName "pg_shadow"
 #define  StatisticRelationName "pg_statistic"
+#define  TableSpaceRelationName "pg_tablespace"
 #define  TypeRelationName "pg_type"
 #define  VersionRelationName "pg_version"
 #define  AttrDefaultRelationName "pg_attrdef"
diff --git a/src/include/catalog/catversion.h b/src/include/catalog/catversion.h
index 6fd25b54f7a..3ba63bbdea3 100644
--- a/src/include/catalog/catversion.h
+++ b/src/include/catalog/catversion.h
@@ -37,7 +37,7 @@
  * Portions Copyright (c) 1996-2003, PostgreSQL Global Development Group
  * Portions Copyright (c) 1994, Regents of the University of California
  *
- * $PostgreSQL: pgsql/src/include/catalog/catversion.h,v 1.236 2004/06/16 01:26:49 tgl Exp $
+ * $PostgreSQL: pgsql/src/include/catalog/catversion.h,v 1.237 2004/06/18 06:14:05 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -53,6 +53,6 @@
  */
 
 /*							yyyymmddN */
-#define CATALOG_VERSION_NO	200406151
+#define CATALOG_VERSION_NO	200406171
 
 #endif
diff --git a/src/include/catalog/heap.h b/src/include/catalog/heap.h
index 2913d53e521..4ead9728df0 100644
--- a/src/include/catalog/heap.h
+++ b/src/include/catalog/heap.h
@@ -7,7 +7,7 @@
  * Portions Copyright (c) 1996-2003, PostgreSQL Global Development Group
  * Portions Copyright (c) 1994, Regents of the University of California
  *
- * $PostgreSQL: pgsql/src/include/catalog/heap.h,v 1.66 2004/05/05 04:48:47 tgl Exp $
+ * $PostgreSQL: pgsql/src/include/catalog/heap.h,v 1.67 2004/06/18 06:14:05 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -37,6 +37,7 @@ typedef struct CookedConstraint
 
 extern Relation heap_create(const char *relname,
 			Oid relnamespace,
+			Oid reltablespace,
 			TupleDesc tupDesc,
 			bool shared_relation,
 			bool storage_create,
@@ -46,6 +47,7 @@ extern void heap_storage_create(Relation rel);
 
 extern Oid heap_create_with_catalog(const char *relname,
 						 Oid relnamespace,
+						 Oid reltablespace,
 						 TupleDesc tupdesc,
 						 char relkind,
 						 bool shared_relation,
diff --git a/src/include/catalog/index.h b/src/include/catalog/index.h
index 841387effcc..be27bb98eb6 100644
--- a/src/include/catalog/index.h
+++ b/src/include/catalog/index.h
@@ -7,7 +7,7 @@
  * Portions Copyright (c) 1996-2003, PostgreSQL Global Development Group
  * Portions Copyright (c) 1994, Regents of the University of California
  *
- * $PostgreSQL: pgsql/src/include/catalog/index.h,v 1.56 2004/05/08 00:34:49 tgl Exp $
+ * $PostgreSQL: pgsql/src/include/catalog/index.h,v 1.57 2004/06/18 06:14:05 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -34,6 +34,7 @@ extern Oid index_create(Oid heapRelationId,
 			 const char *indexRelationName,
 			 IndexInfo *indexInfo,
 			 Oid accessMethodObjectId,
+			 Oid tableSpaceId,
 			 Oid *classObjectId,
 			 bool primary,
 			 bool isconstraint,
diff --git a/src/include/catalog/indexing.h b/src/include/catalog/indexing.h
index 33c0e0f6a2f..c6d2842c78e 100644
--- a/src/include/catalog/indexing.h
+++ b/src/include/catalog/indexing.h
@@ -8,7 +8,7 @@
  * Portions Copyright (c) 1996-2003, PostgreSQL Global Development Group
  * Portions Copyright (c) 1994, Regents of the University of California
  *
- * $PostgreSQL: pgsql/src/include/catalog/indexing.h,v 1.81 2003/11/29 22:40:58 pgsql Exp $
+ * $PostgreSQL: pgsql/src/include/catalog/indexing.h,v 1.82 2004/06/18 06:14:06 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -70,6 +70,8 @@
 #define ShadowNameIndex				"pg_shadow_usename_index"
 #define ShadowSysidIndex			"pg_shadow_usesysid_index"
 #define StatisticRelidAttnumIndex	"pg_statistic_relid_att_index"
+#define TablespaceNameIndex			"pg_tablespace_spcname_index"
+#define TablespaceOidIndex			"pg_tablespace_oid_index"
 #define TriggerConstrNameIndex		"pg_trigger_tgconstrname_index"
 #define TriggerConstrRelidIndex		"pg_trigger_tgconstrrelid_index"
 #define TriggerRelidNameIndex		"pg_trigger_tgrelid_tgname_index"
@@ -166,6 +168,8 @@ DECLARE_UNIQUE_INDEX(pg_rewrite_rel_rulename_index on pg_rewrite using btree(ev_
 DECLARE_UNIQUE_INDEX(pg_shadow_usename_index on pg_shadow using btree(usename name_ops));
 DECLARE_UNIQUE_INDEX(pg_shadow_usesysid_index on pg_shadow using btree(usesysid int4_ops));
 DECLARE_UNIQUE_INDEX(pg_statistic_relid_att_index on pg_statistic using btree(starelid oid_ops, staattnum int2_ops));
+DECLARE_UNIQUE_INDEX(pg_tablespace_oid_index on pg_tablespace using btree(oid oid_ops));
+DECLARE_UNIQUE_INDEX(pg_tablespace_spcname_index on pg_tablespace using btree(spcname name_ops));
 /* This following index is not used for a cache and is not unique */
 DECLARE_INDEX(pg_trigger_tgconstrname_index on pg_trigger using btree(tgconstrname name_ops));
 /* This following index is not used for a cache and is not unique */
diff --git a/src/include/catalog/pg_attribute.h b/src/include/catalog/pg_attribute.h
index 896a06ada86..c3f7ead3af4 100644
--- a/src/include/catalog/pg_attribute.h
+++ b/src/include/catalog/pg_attribute.h
@@ -8,7 +8,7 @@
  * Portions Copyright (c) 1996-2003, PostgreSQL Global Development Group
  * Portions Copyright (c) 1994, Regents of the University of California
  *
- * $PostgreSQL: pgsql/src/include/catalog/pg_attribute.h,v 1.109 2004/04/01 21:28:45 tgl Exp $
+ * $PostgreSQL: pgsql/src/include/catalog/pg_attribute.h,v 1.110 2004/06/18 06:14:06 tgl Exp $
  *
  * NOTES
  *	  the genbki.sh script reads this file and generates .bki
@@ -218,7 +218,7 @@ typedef FormData_pg_attribute *Form_pg_attribute;
  */
 
 /* ----------------
- *		pg_type schema
+ *		pg_type
  * ----------------
  */
 #define Schema_pg_type \
@@ -290,8 +290,7 @@ DATA(insert ( 1262 datallowconn		16 -1 1   5 0 -1 -1 t p c t f f t 0));
 DATA(insert ( 1262 datlastsysoid	26 -1 4   6 0 -1 -1 t p i t f f t 0));
 DATA(insert ( 1262 datvacuumxid		28 -1 4   7 0 -1 -1 t p i t f f t 0));
 DATA(insert ( 1262 datfrozenxid		28 -1 4   8 0 -1 -1 t p i t f f t 0));
-/* do not mark datpath as toastable; GetRawDatabaseInfo won't cope */
-DATA(insert ( 1262 datpath			25 -1 -1  9 0 -1 -1 f p i t f f t 0));
+DATA(insert ( 1262 dattablespace	26 -1 4   9 0 -1 -1 t p i t f f t 0));
 DATA(insert ( 1262 datconfig	  1009 -1 -1 10 1 -1 -1 f x i f f f t 0));
 DATA(insert ( 1262 datacl		  1034 -1 -1 11 1 -1 -1 f x i f f f t 0));
 DATA(insert ( 1262 ctid				27 0  6  -1 0 -1 -1 f p s t f f t 0));
@@ -442,24 +441,25 @@ DATA(insert ( 1249 tableoid			26 0  4  -7 0 -1 -1 t p i t f f t 0));
 { 1259, {"relowner"},	   23, -1,	4,	4, 0, -1, -1, true, 'p', 'i', true, false, false, true, 0 }, \
 { 1259, {"relam"},		   26, -1,	4,	5, 0, -1, -1, true, 'p', 'i', true, false, false, true, 0 }, \
 { 1259, {"relfilenode"},   26, -1,	4,	6, 0, -1, -1, true, 'p', 'i', true, false, false, true, 0 }, \
-{ 1259, {"relpages"},	   23, -1,	4,	7, 0, -1, -1, true, 'p', 'i', true, false, false, true, 0 }, \
-{ 1259, {"reltuples"},	   700, -1,	4,	8, 0, -1, -1, false, 'p', 'i', true, false, false, true, 0 }, \
-{ 1259, {"reltoastrelid"}, 26, -1,	4,	9, 0, -1, -1, true, 'p', 'i', true, false, false, true, 0 }, \
-{ 1259, {"reltoastidxid"}, 26, -1,	4, 10, 0, -1, -1, true, 'p', 'i', true, false, false, true, 0 }, \
-{ 1259, {"relhasindex"},   16, -1,	1, 11, 0, -1, -1, true, 'p', 'c', true, false, false, true, 0 }, \
-{ 1259, {"relisshared"},   16, -1,	1, 12, 0, -1, -1, true, 'p', 'c', true, false, false, true, 0 }, \
-{ 1259, {"relkind"},	   18, -1,	1, 13, 0, -1, -1, true, 'p', 'c', true, false, false, true, 0 }, \
-{ 1259, {"relnatts"},	   21, -1,	2, 14, 0, -1, -1, true, 'p', 's', true, false, false, true, 0 }, \
-{ 1259, {"relchecks"},	   21, -1,	2, 15, 0, -1, -1, true, 'p', 's', true, false, false, true, 0 }, \
-{ 1259, {"reltriggers"},   21, -1,	2, 16, 0, -1, -1, true, 'p', 's', true, false, false, true, 0 }, \
-{ 1259, {"relukeys"},	   21, -1,	2, 17, 0, -1, -1, true, 'p', 's', true, false, false, true, 0 }, \
-{ 1259, {"relfkeys"},	   21, -1,	2, 18, 0, -1, -1, true, 'p', 's', true, false, false, true, 0 }, \
-{ 1259, {"relrefs"},	   21, -1,	2, 19, 0, -1, -1, true, 'p', 's', true, false, false, true, 0 }, \
-{ 1259, {"relhasoids"},    16, -1,	1, 20, 0, -1, -1, true, 'p', 'c', true, false, false, true, 0 }, \
-{ 1259, {"relhaspkey"},    16, -1,	1, 21, 0, -1, -1, true, 'p', 'c', true, false, false, true, 0 }, \
-{ 1259, {"relhasrules"},   16, -1,	1, 22, 0, -1, -1, true, 'p', 'c', true, false, false, true, 0 }, \
-{ 1259, {"relhassubclass"},16, -1,	1, 23, 0, -1, -1, true, 'p', 'c', true, false, false, true, 0 }, \
-{ 1259, {"relacl"},		 1034, -1, -1, 24, 1, -1, -1, false, 'x', 'i', false, false, false, true, 0 }
+{ 1259, {"reltablespace"}, 26, -1,	4,	7, 0, -1, -1, true, 'p', 'i', true, false, false, true, 0 }, \
+{ 1259, {"relpages"},	   23, -1,	4,	8, 0, -1, -1, true, 'p', 'i', true, false, false, true, 0 }, \
+{ 1259, {"reltuples"},	   700, -1,	4,	9, 0, -1, -1, false, 'p', 'i', true, false, false, true, 0 }, \
+{ 1259, {"reltoastrelid"}, 26, -1,	4, 10, 0, -1, -1, true, 'p', 'i', true, false, false, true, 0 }, \
+{ 1259, {"reltoastidxid"}, 26, -1,	4, 11, 0, -1, -1, true, 'p', 'i', true, false, false, true, 0 }, \
+{ 1259, {"relhasindex"},   16, -1,	1, 12, 0, -1, -1, true, 'p', 'c', true, false, false, true, 0 }, \
+{ 1259, {"relisshared"},   16, -1,	1, 13, 0, -1, -1, true, 'p', 'c', true, false, false, true, 0 }, \
+{ 1259, {"relkind"},	   18, -1,	1, 14, 0, -1, -1, true, 'p', 'c', true, false, false, true, 0 }, \
+{ 1259, {"relnatts"},	   21, -1,	2, 15, 0, -1, -1, true, 'p', 's', true, false, false, true, 0 }, \
+{ 1259, {"relchecks"},	   21, -1,	2, 16, 0, -1, -1, true, 'p', 's', true, false, false, true, 0 }, \
+{ 1259, {"reltriggers"},   21, -1,	2, 17, 0, -1, -1, true, 'p', 's', true, false, false, true, 0 }, \
+{ 1259, {"relukeys"},	   21, -1,	2, 18, 0, -1, -1, true, 'p', 's', true, false, false, true, 0 }, \
+{ 1259, {"relfkeys"},	   21, -1,	2, 19, 0, -1, -1, true, 'p', 's', true, false, false, true, 0 }, \
+{ 1259, {"relrefs"},	   21, -1,	2, 20, 0, -1, -1, true, 'p', 's', true, false, false, true, 0 }, \
+{ 1259, {"relhasoids"},    16, -1,	1, 21, 0, -1, -1, true, 'p', 'c', true, false, false, true, 0 }, \
+{ 1259, {"relhaspkey"},    16, -1,	1, 22, 0, -1, -1, true, 'p', 'c', true, false, false, true, 0 }, \
+{ 1259, {"relhasrules"},   16, -1,	1, 23, 0, -1, -1, true, 'p', 'c', true, false, false, true, 0 }, \
+{ 1259, {"relhassubclass"},16, -1,	1, 24, 0, -1, -1, true, 'p', 'c', true, false, false, true, 0 }, \
+{ 1259, {"relacl"},		 1034, -1, -1, 25, 1, -1, -1, false, 'x', 'i', false, false, false, true, 0 }
 
 DATA(insert ( 1259 relname			19 -1 NAMEDATALEN	1 0 -1 -1 f p i t f f t 0));
 DATA(insert ( 1259 relnamespace		26 -1 4   2 0 -1 -1 t p i t f f t 0));
@@ -467,24 +467,25 @@ DATA(insert ( 1259 reltype			26 -1 4   3 0 -1 -1 t p i t f f t 0));
 DATA(insert ( 1259 relowner			23 -1 4   4 0 -1 -1 t p i t f f t 0));
 DATA(insert ( 1259 relam			26 -1 4   5 0 -1 -1 t p i t f f t 0));
 DATA(insert ( 1259 relfilenode		26 -1 4   6 0 -1 -1 t p i t f f t 0));
-DATA(insert ( 1259 relpages			23 -1 4   7 0 -1 -1 t p i t f f t 0));
-DATA(insert ( 1259 reltuples	   700 -1 4   8 0 -1 -1 f p i t f f t 0));
-DATA(insert ( 1259 reltoastrelid	26 -1 4   9 0 -1 -1 t p i t f f t 0));
-DATA(insert ( 1259 reltoastidxid	26 -1 4  10 0 -1 -1 t p i t f f t 0));
-DATA(insert ( 1259 relhasindex		16 -1 1  11 0 -1 -1 t p c t f f t 0));
-DATA(insert ( 1259 relisshared		16 -1 1  12 0 -1 -1 t p c t f f t 0));
-DATA(insert ( 1259 relkind			18 -1 1  13 0 -1 -1 t p c t f f t 0));
-DATA(insert ( 1259 relnatts			21 -1 2  14 0 -1 -1 t p s t f f t 0));
-DATA(insert ( 1259 relchecks		21 -1 2  15 0 -1 -1 t p s t f f t 0));
-DATA(insert ( 1259 reltriggers		21 -1 2  16 0 -1 -1 t p s t f f t 0));
-DATA(insert ( 1259 relukeys			21 -1 2  17 0 -1 -1 t p s t f f t 0));
-DATA(insert ( 1259 relfkeys			21 -1 2  18 0 -1 -1 t p s t f f t 0));
-DATA(insert ( 1259 relrefs			21 -1 2  19 0 -1 -1 t p s t f f t 0));
-DATA(insert ( 1259 relhasoids		16 -1 1  20 0 -1 -1 t p c t f f t 0));
-DATA(insert ( 1259 relhaspkey		16 -1 1  21 0 -1 -1 t p c t f f t 0));
-DATA(insert ( 1259 relhasrules		16 -1 1  22 0 -1 -1 t p c t f f t 0));
-DATA(insert ( 1259 relhassubclass	16 -1 1  23 0 -1 -1 t p c t f f t 0));
-DATA(insert ( 1259 relacl		  1034 -1 -1 24 1 -1 -1 f x i f f f t 0));
+DATA(insert ( 1259 reltablespace	26 -1 4   7 0 -1 -1 t p i t f f t 0));
+DATA(insert ( 1259 relpages			23 -1 4   8 0 -1 -1 t p i t f f t 0));
+DATA(insert ( 1259 reltuples	   700 -1 4   9 0 -1 -1 f p i t f f t 0));
+DATA(insert ( 1259 reltoastrelid	26 -1 4  10 0 -1 -1 t p i t f f t 0));
+DATA(insert ( 1259 reltoastidxid	26 -1 4  11 0 -1 -1 t p i t f f t 0));
+DATA(insert ( 1259 relhasindex		16 -1 1  12 0 -1 -1 t p c t f f t 0));
+DATA(insert ( 1259 relisshared		16 -1 1  13 0 -1 -1 t p c t f f t 0));
+DATA(insert ( 1259 relkind			18 -1 1  14 0 -1 -1 t p c t f f t 0));
+DATA(insert ( 1259 relnatts			21 -1 2  15 0 -1 -1 t p s t f f t 0));
+DATA(insert ( 1259 relchecks		21 -1 2  16 0 -1 -1 t p s t f f t 0));
+DATA(insert ( 1259 reltriggers		21 -1 2  17 0 -1 -1 t p s t f f t 0));
+DATA(insert ( 1259 relukeys			21 -1 2  18 0 -1 -1 t p s t f f t 0));
+DATA(insert ( 1259 relfkeys			21 -1 2  19 0 -1 -1 t p s t f f t 0));
+DATA(insert ( 1259 relrefs			21 -1 2  20 0 -1 -1 t p s t f f t 0));
+DATA(insert ( 1259 relhasoids		16 -1 1  21 0 -1 -1 t p c t f f t 0));
+DATA(insert ( 1259 relhaspkey		16 -1 1  22 0 -1 -1 t p c t f f t 0));
+DATA(insert ( 1259 relhasrules		16 -1 1  23 0 -1 -1 t p c t f f t 0));
+DATA(insert ( 1259 relhassubclass	16 -1 1  24 0 -1 -1 t p c t f f t 0));
+DATA(insert ( 1259 relacl		  1034 -1 -1 25 1 -1 -1 f x i f f f t 0));
 DATA(insert ( 1259 ctid				27 0  6  -1 0 -1 -1 f p s t f f t 0));
 DATA(insert ( 1259 oid				26 0  4  -2 0 -1 -1 t p i t f f t 0));
 DATA(insert ( 1259 xmin				28 0  4  -3 0 -1 -1 t p i t f f t 0));
@@ -494,6 +495,23 @@ DATA(insert ( 1259 cmax				29 0  4  -6 0 -1 -1 t p i t f f t 0));
 DATA(insert ( 1259 tableoid			26 0  4  -7 0 -1 -1 t p i t f f t 0));
 
 /* ----------------
+ *		pg_tablespace
+ * ----------------
+ */
+
+DATA(insert ( 1213 spcname			19 -1 NAMEDATALEN 1 0 -1 -1 f p i t f f t 0));
+DATA(insert ( 1213 spcowner			23 -1 4  2 0  -1 -1 t p i t f f t 0));
+DATA(insert ( 1213 spclocation		25 -1 -1 3 0  -1 -1 f x i t f f t 0));
+DATA(insert ( 1213 spcacl		  1034 -1 -1 4 1  -1 -1 f x i f f f t 0));
+DATA(insert ( 1213 ctid				27 0  6  -1 0 -1 -1 f p s t f f t 0));
+DATA(insert ( 1213 oid				26 0  4  -2 0 -1 -1 t p i t f f t 0));
+DATA(insert ( 1213 xmin				28 0  4  -3 0 -1 -1 t p i t f f t 0));
+DATA(insert ( 1213 cmin				29 0  4  -4 0 -1 -1 t p i t f f t 0));
+DATA(insert ( 1213 xmax				28 0  4  -5 0 -1 -1 t p i t f f t 0));
+DATA(insert ( 1213 cmax				29 0  4  -6 0 -1 -1 t p i t f f t 0));
+DATA(insert ( 1213 tableoid			26 0  4  -7 0 -1 -1 t p i t f f t 0));
+
+/* ----------------
  *		pg_xactlock - this is not a real relation, but is a placeholder
  *				  to allow a relation OID to be used for transaction
  *				  waits.  We need a pg_xactlock entry in pg_class only to
diff --git a/src/include/catalog/pg_class.h b/src/include/catalog/pg_class.h
index 8f044b0459a..bebbc76fa58 100644
--- a/src/include/catalog/pg_class.h
+++ b/src/include/catalog/pg_class.h
@@ -8,7 +8,7 @@
  * Portions Copyright (c) 1996-2003, PostgreSQL Global Development Group
  * Portions Copyright (c) 1994, Regents of the University of California
  *
- * $PostgreSQL: pgsql/src/include/catalog/pg_class.h,v 1.81 2004/04/01 21:28:45 tgl Exp $
+ * $PostgreSQL: pgsql/src/include/catalog/pg_class.h,v 1.82 2004/06/18 06:14:06 tgl Exp $
  *
  * NOTES
  *	  the genbki.sh script reads this file and generates .bki
@@ -48,6 +48,7 @@ CATALOG(pg_class) BOOTSTRAP
 	int4		relowner;		/* class owner */
 	Oid			relam;			/* index access method; 0 if not an index */
 	Oid			relfilenode;	/* identifier of physical storage file */
+	Oid			reltablespace;	/* identifier of table space for relation */
 	int4		relpages;		/* # of blocks (not always up-to-date) */
 	float4		reltuples;		/* # of tuples (not always up-to-date) */
 	Oid			reltoastrelid;	/* OID of toast table; 0 if none */
@@ -100,32 +101,33 @@ typedef FormData_pg_class *Form_pg_class;
  *		relacl field.  This is a kluge.
  * ----------------
  */
-#define Natts_pg_class_fixed			23
-#define Natts_pg_class					24
+#define Natts_pg_class_fixed			24
+#define Natts_pg_class					25
 #define Anum_pg_class_relname			1
 #define Anum_pg_class_relnamespace		2
 #define Anum_pg_class_reltype			3
 #define Anum_pg_class_relowner			4
 #define Anum_pg_class_relam				5
 #define Anum_pg_class_relfilenode		6
-#define Anum_pg_class_relpages			7
-#define Anum_pg_class_reltuples			8
-#define Anum_pg_class_reltoastrelid		9
-#define Anum_pg_class_reltoastidxid		10
-#define Anum_pg_class_relhasindex		11
-#define Anum_pg_class_relisshared		12
-#define Anum_pg_class_relkind			13
-#define Anum_pg_class_relnatts			14
-#define Anum_pg_class_relchecks			15
-#define Anum_pg_class_reltriggers		16
-#define Anum_pg_class_relukeys			17
-#define Anum_pg_class_relfkeys			18
-#define Anum_pg_class_relrefs			19
-#define Anum_pg_class_relhasoids		20
-#define Anum_pg_class_relhaspkey		21
-#define Anum_pg_class_relhasrules		22
-#define Anum_pg_class_relhassubclass	23
-#define Anum_pg_class_relacl			24
+#define Anum_pg_class_reltablespace		7
+#define Anum_pg_class_relpages			8
+#define Anum_pg_class_reltuples			9
+#define Anum_pg_class_reltoastrelid		10
+#define Anum_pg_class_reltoastidxid		11
+#define Anum_pg_class_relhasindex		12
+#define Anum_pg_class_relisshared		13
+#define Anum_pg_class_relkind			14
+#define Anum_pg_class_relnatts			15
+#define Anum_pg_class_relchecks			16
+#define Anum_pg_class_reltriggers		17
+#define Anum_pg_class_relukeys			18
+#define Anum_pg_class_relfkeys			19
+#define Anum_pg_class_relrefs			20
+#define Anum_pg_class_relhasoids		21
+#define Anum_pg_class_relhaspkey		22
+#define Anum_pg_class_relhasrules		23
+#define Anum_pg_class_relhassubclass	24
+#define Anum_pg_class_relacl			25
 
 /* ----------------
  *		initial contents of pg_class
@@ -134,21 +136,23 @@ typedef FormData_pg_class *Form_pg_class;
  * ----------------
  */
 
-DATA(insert OID = 1247 (  pg_type		PGNSP 71 PGUID 0 1247 0 0 0 0 f f r 23 0 0 0 0 0 t f f f _null_ ));
+DATA(insert OID = 1247 (  pg_type		PGNSP 71 PGUID 0 1247 0 0 0 0 0 f f r 23 0 0 0 0 0 t f f f _null_ ));
 DESCR("");
-DATA(insert OID = 1249 (  pg_attribute	PGNSP 75 PGUID 0 1249 0 0 0 0 f f r 17 0 0 0 0 0 f f f f _null_ ));
+DATA(insert OID = 1249 (  pg_attribute	PGNSP 75 PGUID 0 1249 0 0 0 0 0 f f r 17 0 0 0 0 0 f f f f _null_ ));
 DESCR("");
-DATA(insert OID = 1255 (  pg_proc		PGNSP 81 PGUID 0 1255 0 0 0 0 f f r 16 0 0 0 0 0 t f f f _null_ ));
+DATA(insert OID = 1255 (  pg_proc		PGNSP 81 PGUID 0 1255 0 0 0 0 0 f f r 16 0 0 0 0 0 t f f f _null_ ));
 DESCR("");
-DATA(insert OID = 1259 (  pg_class		PGNSP 83 PGUID 0 1259 0 0 0 0 f f r 24 0 0 0 0 0 t f f f _null_ ));
+DATA(insert OID = 1259 (  pg_class		PGNSP 83 PGUID 0 1259 0 0 0 0 0 f f r 25 0 0 0 0 0 t f f f _null_ ));
 DESCR("");
-DATA(insert OID = 1260 (  pg_shadow		PGNSP 86 PGUID 0 1260 0 0 0 0 f t r 8  0 0 0 0 0 f f f f _null_ ));
+DATA(insert OID = 1260 (  pg_shadow		PGNSP 86 PGUID 0 1260 1664 0 0 0 0 f t r 8  0 0 0 0 0 f f f f _null_ ));
 DESCR("");
-DATA(insert OID = 1261 (  pg_group		PGNSP 87 PGUID 0 1261 0 0 0 0 f t r 3  0 0 0 0 0 f f f f _null_ ));
+DATA(insert OID = 1261 (  pg_group		PGNSP 87 PGUID 0 1261 1664 0 0 0 0 f t r 3  0 0 0 0 0 f f f f _null_ ));
 DESCR("");
-DATA(insert OID = 1262 (  pg_database	PGNSP 88 PGUID 0 1262 0 0 0 0 f t r 11	0 0 0 0 0 t f f f _null_ ));
+DATA(insert OID = 1262 (  pg_database	PGNSP 88 PGUID 0 1262 1664 0 0 0 0 f t r 11 0 0 0 0 0 t f f f _null_ ));
 DESCR("");
-DATA(insert OID = 376  (  pg_xactlock	PGNSP  0 PGUID 0	0 0 0 0 0 f t s 1  0 0 0 0 0 f f f f _null_ ));
+DATA(insert OID = 1213 (  pg_tablespace	PGNSP 90 PGUID 0 1213 1664 0 0 0 0 f t r 4  0 0 0 0 0 t f f f _null_ ));
+DESCR("");
+DATA(insert OID = 376  (  pg_xactlock	PGNSP  0 PGUID 0 0 1664 0 0 0 0 f t s 1  0 0 0 0 0 f f f f _null_ ));
 DESCR("");
 
 #define RelOid_pg_type			1247
@@ -158,6 +162,7 @@ DESCR("");
 #define RelOid_pg_shadow		1260
 #define RelOid_pg_group			1261
 #define RelOid_pg_database		1262
+#define RelOid_pg_tablespace	1213
 
 /* Xact lock pseudo-table */
 #define XactLockTableId			376
diff --git a/src/include/catalog/pg_database.h b/src/include/catalog/pg_database.h
index 226c5c2f99c..9504f1733b6 100644
--- a/src/include/catalog/pg_database.h
+++ b/src/include/catalog/pg_database.h
@@ -8,7 +8,7 @@
  * Portions Copyright (c) 1996-2003, PostgreSQL Global Development Group
  * Portions Copyright (c) 1994, Regents of the University of California
  *
- * $PostgreSQL: pgsql/src/include/catalog/pg_database.h,v 1.31 2004/02/10 01:55:26 tgl Exp $
+ * $PostgreSQL: pgsql/src/include/catalog/pg_database.h,v 1.32 2004/06/18 06:14:06 tgl Exp $
  *
  * NOTES
  *	  the genbki.sh script reads this file and generates .bki
@@ -41,7 +41,7 @@ CATALOG(pg_database) BOOTSTRAP BKI_SHARED_RELATION
 	Oid			datlastsysoid;	/* highest OID to consider a system OID */
 	TransactionId datvacuumxid; /* all XIDs before this are vacuumed */
 	TransactionId datfrozenxid; /* all XIDs before this are frozen */
-	text		datpath;		/* default database location (VAR LENGTH) */
+	Oid			dattablespace;	/* default table space for this DB */
 	text		datconfig[1];	/* database-specific GUC (VAR LENGTH) */
 	aclitem		datacl[1];		/* access permissions (VAR LENGTH) */
 } FormData_pg_database;
@@ -66,11 +66,11 @@ typedef FormData_pg_database *Form_pg_database;
 #define Anum_pg_database_datlastsysoid	6
 #define Anum_pg_database_datvacuumxid	7
 #define Anum_pg_database_datfrozenxid	8
-#define Anum_pg_database_datpath		9
+#define Anum_pg_database_dattablespace	9
 #define Anum_pg_database_datconfig		10
 #define Anum_pg_database_datacl			11
 
-DATA(insert OID = 1 (  template1 PGUID ENCODING t t 0 0 0 "" _null_ _null_ ));
+DATA(insert OID = 1 (  template1 PGUID ENCODING t t 0 0 0 1663 _null_ _null_ ));
 DESCR("Default template database");
 #define TemplateDbOid			1
 
diff --git a/src/include/catalog/pg_namespace.h b/src/include/catalog/pg_namespace.h
index 68d8dd28fa6..63a4c708378 100644
--- a/src/include/catalog/pg_namespace.h
+++ b/src/include/catalog/pg_namespace.h
@@ -8,7 +8,7 @@
  * Portions Copyright (c) 1996-2003, PostgreSQL Global Development Group
  * Portions Copyright (c) 1994, Regents of the University of California
  *
- * $PostgreSQL: pgsql/src/include/catalog/pg_namespace.h,v 1.11 2003/11/29 22:40:58 pgsql Exp $
+ * $PostgreSQL: pgsql/src/include/catalog/pg_namespace.h,v 1.12 2004/06/18 06:14:06 tgl Exp $
  *
  * NOTES
  *	  the genbki.sh script reads this file and generates .bki
@@ -40,6 +40,7 @@ CATALOG(pg_namespace)
 {
 	NameData	nspname;
 	int4		nspowner;
+	Oid			nsptablespace;	/* default table space for name space */
 	aclitem		nspacl[1];		/* VARIABLE LENGTH FIELD */
 } FormData_pg_namespace;
 
@@ -55,10 +56,11 @@ typedef FormData_pg_namespace *Form_pg_namespace;
  * ----------------
  */
 
-#define Natts_pg_namespace				3
+#define Natts_pg_namespace				4
 #define Anum_pg_namespace_nspname		1
 #define Anum_pg_namespace_nspowner		2
-#define Anum_pg_namespace_nspacl		3
+#define Anum_pg_namespace_nsptablespace	3
+#define Anum_pg_namespace_nspacl		4
 
 
 /* ----------------
@@ -66,13 +68,13 @@ typedef FormData_pg_namespace *Form_pg_namespace;
  * ---------------
  */
 
-DATA(insert OID = 11 ( "pg_catalog" PGUID _null_ ));
+DATA(insert OID = 11 ( "pg_catalog" PGUID 0 _null_ ));
 DESCR("System catalog schema");
 #define PG_CATALOG_NAMESPACE 11
-DATA(insert OID = 99 ( "pg_toast" PGUID _null_ ));
+DATA(insert OID = 99 ( "pg_toast" PGUID 0 _null_ ));
 DESCR("Reserved schema for TOAST tables");
 #define PG_TOAST_NAMESPACE 99
-DATA(insert OID = 2200 ( "public" PGUID _null_ ));
+DATA(insert OID = 2200 ( "public" PGUID 0 _null_ ));
 DESCR("Standard public schema");
 #define PG_PUBLIC_NAMESPACE 2200
 
@@ -80,6 +82,7 @@ DESCR("Standard public schema");
 /*
  * prototypes for functions in pg_namespace.c
  */
-extern Oid	NamespaceCreate(const char *nspName, int32 ownerSysId);
+extern Oid	NamespaceCreate(const char *nspName, int32 ownerSysId,
+							Oid nspTablespace);
 
 #endif   /* PG_NAMESPACE_H */
diff --git a/src/include/catalog/pg_tablespace.h b/src/include/catalog/pg_tablespace.h
new file mode 100644
index 00000000000..f05d3657704
--- /dev/null
+++ b/src/include/catalog/pg_tablespace.h
@@ -0,0 +1,66 @@
+/*-------------------------------------------------------------------------
+ *
+ * pg_tablespace.h
+ *	  definition of the system "tablespace" relation (pg_tablespace)
+ *	  along with the relation's initial contents.
+ *
+ *
+ * Portions Copyright (c) 1996-2003, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * $PostgreSQL: pgsql/src/include/catalog/pg_tablespace.h,v 1.1 2004/06/18 06:14:06 tgl Exp $
+ *
+ * NOTES
+ *	  the genbki.sh script reads this file and generates .bki
+ *	  information from the DATA() statements.
+ *
+ *-------------------------------------------------------------------------
+ */
+#ifndef PG_TABLESPACE_H
+#define PG_TABLESPACE_H
+
+/* ----------------
+ *		postgres.h contains the system type definitions and the
+ *		CATALOG(), BOOTSTRAP and DATA() sugar words so this file
+ *		can be read by both genbki.sh and the C compiler.
+ * ----------------
+ */
+
+/* ----------------
+ *		pg_tablespace definition.  cpp turns this into
+ *		typedef struct FormData_pg_tablespace
+ * ----------------
+ */
+CATALOG(pg_tablespace) BOOTSTRAP BKI_SHARED_RELATION
+{
+	NameData	spcname;		/* tablespace name */
+	int4		spcowner;		/* sysid of owner */
+	text		spclocation;	/* physical location (VAR LENGTH) */
+	aclitem		spcacl[1];		/* access permissions (VAR LENGTH) */
+} FormData_pg_tablespace;
+
+/* ----------------
+ *		Form_pg_tablespace corresponds to a pointer to a tuple with
+ *		the format of pg_tablespace relation.
+ * ----------------
+ */
+typedef FormData_pg_tablespace *Form_pg_tablespace;
+
+/* ----------------
+ *		compiler constants for pg_tablespace
+ * ----------------
+ */
+
+#define Natts_pg_tablespace				4
+#define Anum_pg_tablespace_spcname		1
+#define Anum_pg_tablespace_spcowner		2
+#define Anum_pg_tablespace_spclocation	3
+#define Anum_pg_tablespace_spcacl		4
+
+DATA(insert OID = 1663 ( default	PGUID "" _null_ ));
+DATA(insert OID = 1664 ( global		PGUID "" _null_ ));
+
+#define DEFAULTTABLESPACE_OID 1663
+#define GLOBALTABLESPACE_OID 1664
+
+#endif   /* PG_TABLESPACE_H */
diff --git a/src/include/catalog/pg_type.h b/src/include/catalog/pg_type.h
index 3b9b1dfd5db..45d92b36277 100644
--- a/src/include/catalog/pg_type.h
+++ b/src/include/catalog/pg_type.h
@@ -8,7 +8,7 @@
  * Portions Copyright (c) 1996-2003, PostgreSQL Global Development Group
  * Portions Copyright (c) 1994, Regents of the University of California
  *
- * $PostgreSQL: pgsql/src/include/catalog/pg_type.h,v 1.153 2004/06/06 19:07:01 tgl Exp $
+ * $PostgreSQL: pgsql/src/include/catalog/pg_type.h,v 1.154 2004/06/18 06:14:06 tgl Exp $
  *
  * NOTES
  *	  the genbki.sh script reads this file and generates .bki
@@ -304,13 +304,14 @@ DATA(insert OID = 30 (	oidvector  PGNSP PGUID INDEX_MAX_KEYS*4 f b t \054 0  26
 DESCR("array of INDEX_MAX_KEYS oids, used in system tables");
 #define OIDVECTOROID	30
 
-DATA(insert OID = 71 (	pg_type		 PGNSP PGUID -1 f c t \054 1247 0 record_in record_out record_recv record_send - d x f 0 -1 0 _null_ _null_ ));
-DATA(insert OID = 75 (	pg_attribute PGNSP PGUID -1 f c t \054 1249 0 record_in record_out record_recv record_send - d x f 0 -1 0 _null_ _null_ ));
-DATA(insert OID = 81 (	pg_proc		 PGNSP PGUID -1 f c t \054 1255 0 record_in record_out record_recv record_send - d x f 0 -1 0 _null_ _null_ ));
-DATA(insert OID = 83 (	pg_class	 PGNSP PGUID -1 f c t \054 1259 0 record_in record_out record_recv record_send - d x f 0 -1 0 _null_ _null_ ));
-DATA(insert OID = 86 (	pg_shadow	 PGNSP PGUID -1 f c t \054 1260 0 record_in record_out record_recv record_send - d x f 0 -1 0 _null_ _null_ ));
-DATA(insert OID = 87 (	pg_group	 PGNSP PGUID -1 f c t \054 1261 0 record_in record_out record_recv record_send - d x f 0 -1 0 _null_ _null_ ));
-DATA(insert OID = 88 (	pg_database  PGNSP PGUID -1 f c t \054 1262 0 record_in record_out record_recv record_send - d x f 0 -1 0 _null_ _null_ ));
+DATA(insert OID = 71 (	pg_type		 	PGNSP PGUID -1 f c t \054 1247 0 record_in record_out record_recv record_send - d x f 0 -1 0 _null_ _null_ ));
+DATA(insert OID = 75 (	pg_attribute 	PGNSP PGUID -1 f c t \054 1249 0 record_in record_out record_recv record_send - d x f 0 -1 0 _null_ _null_ ));
+DATA(insert OID = 81 (	pg_proc		 	PGNSP PGUID -1 f c t \054 1255 0 record_in record_out record_recv record_send - d x f 0 -1 0 _null_ _null_ ));
+DATA(insert OID = 83 (	pg_class	 	PGNSP PGUID -1 f c t \054 1259 0 record_in record_out record_recv record_send - d x f 0 -1 0 _null_ _null_ ));
+DATA(insert OID = 86 (	pg_shadow	 	PGNSP PGUID -1 f c t \054 1260 0 record_in record_out record_recv record_send - d x f 0 -1 0 _null_ _null_ ));
+DATA(insert OID = 87 (	pg_group	 	PGNSP PGUID -1 f c t \054 1261 0 record_in record_out record_recv record_send - d x f 0 -1 0 _null_ _null_ ));
+DATA(insert OID = 88 (	pg_database  	PGNSP PGUID -1 f c t \054 1262 0 record_in record_out record_recv record_send - d x f 0 -1 0 _null_ _null_ ));
+DATA(insert OID = 90 (	pg_tablespace	PGNSP PGUID -1 f c t \054 1213 0 record_in record_out record_recv record_send - d x f 0 -1 0 _null_ _null_ ));
 
 /* OIDS 100 - 199 */
 
diff --git a/src/include/commands/defrem.h b/src/include/commands/defrem.h
index ba7415a582d..8670a41a763 100644
--- a/src/include/commands/defrem.h
+++ b/src/include/commands/defrem.h
@@ -7,7 +7,7 @@
  * Portions Copyright (c) 1996-2003, PostgreSQL Global Development Group
  * Portions Copyright (c) 1994, Regents of the University of California
  *
- * $PostgreSQL: pgsql/src/include/commands/defrem.h,v 1.57 2004/06/10 17:55:59 tgl Exp $
+ * $PostgreSQL: pgsql/src/include/commands/defrem.h,v 1.58 2004/06/18 06:14:08 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -21,6 +21,7 @@
 extern void DefineIndex(RangeVar *heapRelation,
 			char *indexRelationName,
 			char *accessMethodName,
+			char *tableSpaceName,
 			List *attributeList,
 			Expr *predicate,
 			List *rangetable,
diff --git a/src/include/commands/tablespace.h b/src/include/commands/tablespace.h
new file mode 100644
index 00000000000..129413ac14b
--- /dev/null
+++ b/src/include/commands/tablespace.h
@@ -0,0 +1,29 @@
+/*-------------------------------------------------------------------------
+ *
+ * tablespace.h
+ *	  prototypes for tablespace.c.
+ *
+ *
+ * Portions Copyright (c) 1996-2004, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * $PostgreSQL: pgsql/src/include/commands/tablespace.h,v 1.1 2004/06/18 06:14:08 tgl Exp $
+ *
+ *-------------------------------------------------------------------------
+ */
+#ifndef TABLESPACE_H
+#define TABLESPACE_H
+
+#include "nodes/parsenodes.h"
+
+extern void CreateTableSpace(CreateTableSpaceStmt *stmt);
+
+extern void DropTableSpace(DropTableSpaceStmt *stmt);
+
+extern void TablespaceCreateDbspace(Oid spcNode, Oid dbNode);
+
+extern Oid	get_tablespace_oid(const char *tablespacename);
+
+extern char *get_tablespace_name(Oid spc_oid);
+
+#endif   /* TABLESPACE_H */
diff --git a/src/include/miscadmin.h b/src/include/miscadmin.h
index 4561d2fcfb7..0a508861b27 100644
--- a/src/include/miscadmin.h
+++ b/src/include/miscadmin.h
@@ -13,7 +13,7 @@
  * Portions Copyright (c) 1996-2003, PostgreSQL Global Development Group
  * Portions Copyright (c) 1994, Regents of the University of California
  *
- * $PostgreSQL: pgsql/src/include/miscadmin.h,v 1.162 2004/05/29 22:48:22 tgl Exp $
+ * $PostgreSQL: pgsql/src/include/miscadmin.h,v 1.163 2004/06/18 06:14:10 tgl Exp $
  *
  * NOTES
  *	  some of the information in this file should be moved to other files.
@@ -150,6 +150,8 @@ extern char postgres_exec_path[];
  */
 extern DLLIMPORT Oid MyDatabaseId;
 
+extern DLLIMPORT Oid MyDatabaseTableSpace;
+
 /*
  * Date/Time Configuration
  *
@@ -224,7 +226,7 @@ extern void check_stack_depth(void);
 extern char *DatabasePath;
 
 /* in utils/misc/database.c */
-extern void GetRawDatabaseInfo(const char *name, Oid *db_id, char *path);
+extern void GetRawDatabaseInfo(const char *name, Oid *db_id, Oid *db_tablespace);
 
 /* now in utils/init/miscinit.c */
 extern void SetDatabasePath(const char *path);
diff --git a/src/include/nodes/nodes.h b/src/include/nodes/nodes.h
index b719747e762..531b7e6c654 100644
--- a/src/include/nodes/nodes.h
+++ b/src/include/nodes/nodes.h
@@ -7,7 +7,7 @@
  * Portions Copyright (c) 1996-2003, PostgreSQL Global Development Group
  * Portions Copyright (c) 1994, Regents of the University of California
  *
- * $PostgreSQL: pgsql/src/include/nodes/nodes.h,v 1.157 2004/06/09 19:08:18 tgl Exp $
+ * $PostgreSQL: pgsql/src/include/nodes/nodes.h,v 1.158 2004/06/18 06:14:11 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -268,6 +268,8 @@ typedef enum NodeTag
 	T_ExecuteStmt,
 	T_DeallocateStmt,
 	T_DeclareCursorStmt,
+	T_CreateTableSpaceStmt,
+	T_DropTableSpaceStmt,
 	T_AlterDbOwnerStmt,
 
 	T_A_Expr = 800,
diff --git a/src/include/nodes/parsenodes.h b/src/include/nodes/parsenodes.h
index 0da9b379077..4c24fe9e27b 100644
--- a/src/include/nodes/parsenodes.h
+++ b/src/include/nodes/parsenodes.h
@@ -7,7 +7,7 @@
  * Portions Copyright (c) 1996-2003, PostgreSQL Global Development Group
  * Portions Copyright (c) 1994, Regents of the University of California
  *
- * $PostgreSQL: pgsql/src/include/nodes/parsenodes.h,v 1.258 2004/06/09 19:08:18 tgl Exp $
+ * $PostgreSQL: pgsql/src/include/nodes/parsenodes.h,v 1.259 2004/06/18 06:14:11 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -742,6 +742,7 @@ typedef enum ObjectType
 	OBJECT_SCHEMA,
 	OBJECT_SEQUENCE,
 	OBJECT_TABLE,
+	OBJECT_TABLESPACE,
 	OBJECT_TRIGGER,
 	OBJECT_TYPE,
 	OBJECT_USER,
@@ -761,6 +762,7 @@ typedef struct CreateSchemaStmt
 	NodeTag		type;
 	char	   *schemaname;		/* the name of the schema to create */
 	char	   *authid;			/* the owner of the created schema */
+	char	   *tablespacename;	/* default tablespace for schema, or NULL */
 	List	   *schemaElts;		/* schema components (list of parsenodes) */
 } CreateSchemaStmt;
 
@@ -857,7 +859,8 @@ typedef enum GrantObjectType
 	ACL_OBJECT_DATABASE,		/* database */
 	ACL_OBJECT_FUNCTION,		/* function */
 	ACL_OBJECT_LANGUAGE,		/* procedural language */
-	ACL_OBJECT_NAMESPACE		/* namespace */
+	ACL_OBJECT_NAMESPACE,		/* namespace */
+	ACL_OBJECT_TABLESPACE		/* tablespace */
 } GrantObjectType;
 
 typedef struct GrantStmt
@@ -941,6 +944,7 @@ typedef struct CreateStmt
 	List	   *constraints;	/* constraints (list of Constraint nodes) */
 	ContainsOids   hasoids;		/* should it have OIDs? */
 	OnCommitAction oncommit;	/* what do we do at COMMIT? */
+	char	   *tablespacename;	/* table space to use, or NULL */
 } CreateStmt;
 
 /* ----------
@@ -1030,6 +1034,26 @@ typedef struct FkConstraint
 	bool		skip_validation;	/* skip validation of existing rows? */
 } FkConstraint;
 
+
+/* ----------------------
+ *      Create/Drop Table Space Statements
+ * ----------------------
+ */
+
+typedef struct CreateTableSpaceStmt
+{
+	NodeTag		type;
+	char	   *tablespacename;
+	char	   *owner;
+	char	   *location;
+} CreateTableSpaceStmt;
+
+typedef struct DropTableSpaceStmt
+{
+	NodeTag		type;
+	char	   *tablespacename;
+} DropTableSpaceStmt;
+
 /* ----------------------
  *		Create/Drop TRIGGER Statements
  * ----------------------
@@ -1142,6 +1166,7 @@ typedef struct CreateSeqStmt
 	NodeTag		type;
 	RangeVar   *sequence;		/* the sequence to create */
 	List	   *options;
+	char	   *tablespacename;	/* tablespace, or NULL for default */
 } CreateSeqStmt;
 
 typedef struct AlterSeqStmt
@@ -1322,6 +1347,7 @@ typedef struct IndexStmt
 	char	   *idxname;		/* name of new index, or NULL for default */
 	RangeVar   *relation;		/* relation to build index on */
 	char	   *accessMethod;	/* name of access method (eg. btree) */
+	char	   *tableSpace;		/* tablespace, or NULL to use parent's */
 	List	   *indexParams;	/* a list of IndexElem */
 	Node	   *whereClause;	/* qualification (partial-index predicate) */
 	List	   *rangetable;		/* range table for qual and/or
diff --git a/src/include/storage/buf_internals.h b/src/include/storage/buf_internals.h
index 6db4afd58bd..4a8b20908e5 100644
--- a/src/include/storage/buf_internals.h
+++ b/src/include/storage/buf_internals.h
@@ -8,7 +8,7 @@
  * Portions Copyright (c) 1996-2003, PostgreSQL Global Development Group
  * Portions Copyright (c) 1994, Regents of the University of California
  *
- * $PostgreSQL: pgsql/src/include/storage/buf_internals.h,v 1.70 2004/04/21 18:06:29 tgl Exp $
+ * $PostgreSQL: pgsql/src/include/storage/buf_internals.h,v 1.71 2004/06/18 06:14:13 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -53,7 +53,8 @@ typedef struct buftag
 
 #define CLEAR_BUFFERTAG(a) \
 ( \
-	(a).rnode.tblNode = InvalidOid, \
+	(a).rnode.spcNode = InvalidOid, \
+	(a).rnode.dbNode = InvalidOid, \
 	(a).rnode.relNode = InvalidOid, \
 	(a).blockNum = InvalidBlockNumber \
 )
diff --git a/src/include/storage/relfilenode.h b/src/include/storage/relfilenode.h
index 715dc00f7de..d430412e67f 100644
--- a/src/include/storage/relfilenode.h
+++ b/src/include/storage/relfilenode.h
@@ -1,22 +1,60 @@
+/*-------------------------------------------------------------------------
+ *
+ * relfilenode.h
+ *	  Physical access information for relations.
+ *
+ *
+ * Portions Copyright (c) 1996-2003, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * $PostgreSQL: pgsql/src/include/storage/relfilenode.h,v 1.9 2004/06/18 06:14:13 tgl Exp $
+ *
+ *-------------------------------------------------------------------------
+ */
 #ifndef RELFILENODE_H
 #define RELFILENODE_H
 
 /*
- * This is all what we need to know to find relation file.
- * tblNode is identificator of tablespace and because of
- * currently our tablespaces are equal to databases this is
- * database OID. relNode is currently relation OID on creation
- * but may be changed later if required. relNode is stored in
- * pg_class.relfilenode.
+ * RelFileNode must provide all that we need to know to physically access
+ * a relation.
+ *
+ * spcNode identifies the tablespace of the relation.  It corresponds to
+ * pg_tablespace.oid.
+ *
+ * dbNode identifies the database of the relation.  It is zero for
+ * "shared" relations (those common to all databases of a cluster).
+ * Nonzero dbNode values correspond to pg_database.oid.
+ *
+ * relNode identifies the specific relation.  relNode corresponds to
+ * pg_class.relfilenode (NOT pg_class.oid, because we need to be able
+ * to assign new physical files to relations in some situations).
+ * Notice that relNode is only unique within a particular database.
+ *
+ * Note: spcNode must be GLOBALTABLESPACE_OID if and only if dbNode is
+ * zero.  We support shared relations only in the "global" tablespace.
+ *
+ * Note: in pg_class we allow reltablespace == 0 to denote that the
+ * relation is stored in its database's "default" tablespace (as
+ * identified by pg_database.dattablespace).  However this shorthand
+ * is NOT allowed in RelFileNode structs --- the real tablespace ID
+ * must be supplied when setting spcNode.
  */
 typedef struct RelFileNode
 {
-	Oid			tblNode;		/* tablespace */
+	Oid			spcNode;		/* tablespace */
+	Oid			dbNode;			/* database */
 	Oid			relNode;		/* relation */
 } RelFileNode;
 
+/*
+ * Note: RelFileNodeEquals compares relNode first since that is most likely
+ * to be different in two unequal RelFileNodes.  It is probably redundant
+ * to compare spcNode if the other two fields are found equal, but do it
+ * anyway to be sure.
+ */
 #define RelFileNodeEquals(node1, node2) \
 	((node1).relNode == (node2).relNode && \
-	 (node1).tblNode == (node2).tblNode)
+	 (node1).dbNode == (node2).dbNode && \
+	 (node1).spcNode == (node2).spcNode)
 
-#endif   /* RELFILENODE_H */
+#endif /* RELFILENODE_H */
diff --git a/src/include/utils/acl.h b/src/include/utils/acl.h
index f5ac89e2579..f7779be589d 100644
--- a/src/include/utils/acl.h
+++ b/src/include/utils/acl.h
@@ -7,7 +7,7 @@
  * Portions Copyright (c) 1996-2003, PostgreSQL Global Development Group
  * Portions Copyright (c) 1994, Regents of the University of California
  *
- * $PostgreSQL: pgsql/src/include/utils/acl.h,v 1.70 2004/06/01 21:49:22 tgl Exp $
+ * $PostgreSQL: pgsql/src/include/utils/acl.h,v 1.71 2004/06/18 06:14:21 tgl Exp $
  *
  * NOTES
  *	  An ACL array is simply an array of AclItems, representing the union
@@ -184,6 +184,7 @@ typedef ArrayType IdList;
 #define ACL_ALL_RIGHTS_FUNCTION		(ACL_EXECUTE)
 #define ACL_ALL_RIGHTS_LANGUAGE		(ACL_USAGE)
 #define ACL_ALL_RIGHTS_NAMESPACE	(ACL_USAGE|ACL_CREATE)
+#define ACL_ALL_RIGHTS_TABLESPACE	(ACL_CREATE)
 
 /* operation codes for pg_*_aclmask */
 typedef enum
@@ -213,6 +214,7 @@ typedef enum AclObjectKind
 	ACL_KIND_NAMESPACE,			/* pg_namespace */
 	ACL_KIND_OPCLASS,			/* pg_opclass */
 	ACL_KIND_CONVERSION,		/* pg_conversion */
+	ACL_KIND_TABLESPACE,		/* pg_tablespace */
 	MAX_ACL_KIND				/* MUST BE LAST */
 } AclObjectKind;
 
@@ -254,12 +256,15 @@ extern AclMode pg_language_aclmask(Oid lang_oid, AclId userid,
 								   AclMode mask, AclMaskHow how);
 extern AclMode pg_namespace_aclmask(Oid nsp_oid, AclId userid,
 									AclMode mask, AclMaskHow how);
+extern AclMode pg_tablespace_aclmask(Oid spc_oid, AclId userid,
+						            AclMode mask, AclMaskHow how);
 
 extern AclResult pg_class_aclcheck(Oid table_oid, AclId userid, AclMode mode);
 extern AclResult pg_database_aclcheck(Oid db_oid, AclId userid, AclMode mode);
 extern AclResult pg_proc_aclcheck(Oid proc_oid, AclId userid, AclMode mode);
 extern AclResult pg_language_aclcheck(Oid lang_oid, AclId userid, AclMode mode);
 extern AclResult pg_namespace_aclcheck(Oid nsp_oid, AclId userid, AclMode mode);
+extern AclResult pg_tablespace_aclcheck(Oid spc_oid, AclId userid, AclMode mode);
 
 extern void aclcheck_error(AclResult aclerr, AclObjectKind objectkind,
 			   const char *objectname);
@@ -270,6 +275,7 @@ extern bool pg_type_ownercheck(Oid type_oid, AclId userid);
 extern bool pg_oper_ownercheck(Oid oper_oid, AclId userid);
 extern bool pg_proc_ownercheck(Oid proc_oid, AclId userid);
 extern bool pg_namespace_ownercheck(Oid nsp_oid, AclId userid);
+extern bool pg_tablespace_ownercheck(Oid spc_oid, AclId userid);
 extern bool pg_opclass_ownercheck(Oid opc_oid, AclId userid);
 extern bool pg_database_ownercheck(Oid db_oid, AclId userid);
 extern bool pg_conversion_ownercheck(Oid conv_oid, AclId userid);
diff --git a/src/include/utils/lsyscache.h b/src/include/utils/lsyscache.h
index 68fab871bde..10d2104b8d0 100644
--- a/src/include/utils/lsyscache.h
+++ b/src/include/utils/lsyscache.h
@@ -6,7 +6,7 @@
  * Portions Copyright (c) 1996-2003, PostgreSQL Global Development Group
  * Portions Copyright (c) 1994, Regents of the University of California
  *
- * $PostgreSQL: pgsql/src/include/utils/lsyscache.h,v 1.87 2004/06/06 00:41:28 tgl Exp $
+ * $PostgreSQL: pgsql/src/include/utils/lsyscache.h,v 1.88 2004/06/18 06:14:21 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -63,6 +63,7 @@ extern Oid	get_relname_relid(const char *relname, Oid relnamespace);
 extern Oid	get_system_catalog_relid(const char *catname);
 extern char *get_rel_name(Oid relid);
 extern Oid	get_rel_namespace(Oid relid);
+extern Oid	get_rel_tablespace(Oid relid);
 extern Oid	get_rel_type_id(Oid relid);
 extern char get_rel_relkind(Oid relid);
 extern bool get_typisdefined(Oid typid);
@@ -105,6 +106,7 @@ extern void free_attstatsslot(Oid atttype,
 				  Datum *values, int nvalues,
 				  float4 *numbers, int nnumbers);
 extern char *get_namespace_name(Oid nspid);
+extern Oid	get_namespace_tablespace(Oid nspid);
 extern int32 get_usesysid(const char *username);
 
 #define is_array_type(typid)  (get_element_type(typid) != InvalidOid)
diff --git a/src/include/utils/relcache.h b/src/include/utils/relcache.h
index 848d68b2077..da82f4f6137 100644
--- a/src/include/utils/relcache.h
+++ b/src/include/utils/relcache.h
@@ -7,7 +7,7 @@
  * Portions Copyright (c) 1996-2003, PostgreSQL Global Development Group
  * Portions Copyright (c) 1994, Regents of the University of California
  *
- * $PostgreSQL: pgsql/src/include/utils/relcache.h,v 1.39 2004/02/10 01:55:27 tgl Exp $
+ * $PostgreSQL: pgsql/src/include/utils/relcache.h,v 1.40 2004/06/18 06:14:21 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -51,8 +51,9 @@ extern void RelationCacheInitializePhase3(void);
 extern Relation RelationBuildLocalRelation(const char *relname,
 						   Oid relnamespace,
 						   TupleDesc tupDesc,
-						   Oid relid, Oid dbid,
-						   RelFileNode rnode,
+						   Oid relid,
+						   Oid reltablespace,
+						   bool shared_relation,
 						   bool nailit);
 
 /*
diff --git a/src/test/regress/GNUmakefile b/src/test/regress/GNUmakefile
index 61a2f9f790a..1fdd1f1e5c1 100644
--- a/src/test/regress/GNUmakefile
+++ b/src/test/regress/GNUmakefile
@@ -7,7 +7,7 @@
 #
 #
 # IDENTIFICATION
-#    $PostgreSQL: pgsql/src/test/regress/GNUmakefile,v 1.46 2004/03/03 04:22:47 momjian Exp $
+#    $PostgreSQL: pgsql/src/test/regress/GNUmakefile,v 1.47 2004/06/18 06:14:25 tgl Exp $
 #
 #-------------------------------------------------------------------------
 
@@ -67,7 +67,7 @@ all: $(DLOBJS)
 
 # Build test input and expected files
 
-file_list := copy create_function_1 create_function_2 misc constraints
+file_list := copy create_function_1 create_function_2 misc constraints tablespace
 input_files  := $(foreach file, $(file_list), sql/$(file).sql)
 output_files := $(foreach file, $(file_list), expected/$(file).out)
 
@@ -81,10 +81,13 @@ abs_srcdir := $(shell cd $(srcdir) && pwd -W)
 abs_builddir := $(shell pwd -W)
 endif
 
+testtablespace := $(abs_builddir)/testtablespace
+
 
 define sed-command
 sed -e 's,@abs_srcdir@,$(abs_srcdir),g' \
     -e 's,@abs_builddir@,$(abs_builddir),g' \
+    -e 's,@testtablespace@,$(testtablespace),g' \
     -e 's/@DLSUFFIX@/$(DLSUFFIX)/g' $< >$@
 endef
 
@@ -125,9 +128,13 @@ all-spi:
 ##
 
 check: all
+	-rm -rf ./testtablespace
+	mkdir ./testtablespace
 	$(SHELL) ./pg_regress --temp-install --top-builddir=$(top_builddir) --schedule=$(srcdir)/parallel_schedule --multibyte=$(MULTIBYTE) $(MAXCONNOPT)
 
 installcheck: all
+	-rm -rf ./testtablespace
+	mkdir ./testtablespace
 	$(SHELL) ./pg_regress --schedule=$(srcdir)/serial_schedule --multibyte=$(MULTIBYTE)
 
 
@@ -152,6 +159,7 @@ clean distclean maintainer-clean:
 	$(MAKE) -C $(contribdir)/spi clean
 	rm -f $(output_files) $(input_files) $(DLOBJS) regress.o pg_regress
 # things created by various check targets
+	rm -rf testtablespace
 	rm -rf results tmp_check log
 	rm -f regression.diffs regression.out regress.out run_check.out
 ifeq ($(PORTNAME), cygwin)
diff --git a/src/test/regress/expected/sanity_check.out b/src/test/regress/expected/sanity_check.out
index 87e1943e5a2..00abd941ccf 100644
--- a/src/test/regress/expected/sanity_check.out
+++ b/src/test/regress/expected/sanity_check.out
@@ -56,13 +56,14 @@ SELECT relname, relhasindex
  pg_rewrite          | t
  pg_shadow           | t
  pg_statistic        | t
+ pg_tablespace       | t
  pg_trigger          | t
  pg_type             | t
  road                | t
  shighway            | t
  tenk1               | t
  tenk2               | t
-(52 rows)
+(53 rows)
 
 --
 -- another sanity check: every system catalog that has OIDs should have
diff --git a/src/test/regress/input/tablespace.source b/src/test/regress/input/tablespace.source
new file mode 100644
index 00000000000..460d4433a84
--- /dev/null
+++ b/src/test/regress/input/tablespace.source
@@ -0,0 +1,36 @@
+-- create a tablespace we can use
+CREATE TABLESPACE testspace LOCATION '@testtablespace@';
+
+-- create a schema in the tablespace
+CREATE SCHEMA testschema TABLESPACE testspace;
+
+-- sanity check
+SELECT nspname, spcname FROM pg_catalog.pg_tablespace t, pg_catalog.pg_namespace n
+	where n.nsptablespace = t.oid and n.nspname = 'testschema';
+
+-- try a table
+CREATE TABLE testschema.foo (i int);
+SELECT relname, spcname FROM pg_catalog.pg_tablespace t, pg_catalog.pg_class c
+    where c.reltablespace = t.oid AND c.relname = 'foo';
+
+INSERT INTO testschema.foo VALUES(1);
+INSERT INTO testschema.foo VALUES(2);
+
+-- index
+CREATE INDEX foo_idx on testschema.foo(i);
+SELECT relname, spcname FROM pg_catalog.pg_tablespace t, pg_catalog.pg_class c
+    where c.reltablespace = t.oid AND c.relname = 'foo_idx';
+
+-- Will fail with bad path
+CREATE TABLESPACE badspace LOCATION '/no/such/location';
+
+-- No such tablespace
+CREATE TABLE bar (i int) TABLESPACE nosuchspace;
+
+-- Fail, not empty
+DROP TABLESPACE testspace;
+
+DROP SCHEMA testschema CASCADE;
+
+-- Should succeed
+DROP TABLESPACE testspace;
diff --git a/src/test/regress/output/tablespace.source b/src/test/regress/output/tablespace.source
new file mode 100644
index 00000000000..43044a2bcf8
--- /dev/null
+++ b/src/test/regress/output/tablespace.source
@@ -0,0 +1,45 @@
+-- create a tablespace we can use
+CREATE TABLESPACE testspace LOCATION '@testtablespace@';
+-- create a schema in the tablespace
+CREATE SCHEMA testschema TABLESPACE testspace;
+-- sanity check
+SELECT nspname, spcname FROM pg_catalog.pg_tablespace t, pg_catalog.pg_namespace n
+	where n.nsptablespace = t.oid and n.nspname = 'testschema';
+  nspname   |  spcname  
+------------+-----------
+ testschema | testspace
+(1 row)
+
+-- try a table
+CREATE TABLE testschema.foo (i int);
+SELECT relname, spcname FROM pg_catalog.pg_tablespace t, pg_catalog.pg_class c
+    where c.reltablespace = t.oid AND c.relname = 'foo';
+ relname |  spcname  
+---------+-----------
+ foo     | testspace
+(1 row)
+
+INSERT INTO testschema.foo VALUES(1);
+INSERT INTO testschema.foo VALUES(2);
+-- index
+CREATE INDEX foo_idx on testschema.foo(i);
+SELECT relname, spcname FROM pg_catalog.pg_tablespace t, pg_catalog.pg_class c
+    where c.reltablespace = t.oid AND c.relname = 'foo_idx';
+ relname |  spcname  
+---------+-----------
+ foo_idx | testspace
+(1 row)
+
+-- Will fail with bad path
+CREATE TABLESPACE badspace LOCATION '/no/such/location';
+ERROR:  could not set permissions on directory "/no/such/location": No such file or directory
+-- No such tablespace
+CREATE TABLE bar (i int) TABLESPACE nosuchspace;
+ERROR:  tablespace "nosuchspace" does not exist
+-- Fail, not empty
+DROP TABLESPACE testspace;
+ERROR:  tablespace "testspace" is not empty
+DROP SCHEMA testschema CASCADE;
+NOTICE:  drop cascades to table testschema.foo
+-- Should succeed
+DROP TABLESPACE testspace;
diff --git a/src/test/regress/parallel_schedule b/src/test/regress/parallel_schedule
index 035a81543f3..b3a4de3b56f 100644
--- a/src/test/regress/parallel_schedule
+++ b/src/test/regress/parallel_schedule
@@ -16,7 +16,7 @@ test: point lseg box path polygon circle date time timetz timestamp timestamptz
 # Depends on point, lseg, box, path, polygon and circle
 test: geometry
 # Depends on interval, timetz, timestamp, timestamptz, reltime and abstime
-test: horology	
+test: horology
 
 # ----------
 # These four each depend on the previous one
@@ -78,3 +78,6 @@ test: limit plpgsql copy2 temp domain rangefuncs prepare without_oid conversion
 
 # run stats by itself because its delay may be insufficient under heavy load
 test: stats
+
+# run tablespace by itself
+test: tablespace
diff --git a/src/test/regress/serial_schedule b/src/test/regress/serial_schedule
index 1dd7a03cc7a..f8621b404ac 100644
--- a/src/test/regress/serial_schedule
+++ b/src/test/regress/serial_schedule
@@ -1,4 +1,4 @@
-# $PostgreSQL: pgsql/src/test/regress/serial_schedule,v 1.25 2004/06/06 21:20:46 tgl Exp $
+# $PostgreSQL: pgsql/src/test/regress/serial_schedule,v 1.26 2004/06/18 06:14:25 tgl Exp $
 # This should probably be in an order similar to parallel_schedule.
 test: boolean
 test: char
@@ -96,3 +96,4 @@ test: sequence
 test: polymorphism
 test: rowtypes
 test: stats
+test: tablespace
author	Tom Lane <tgl@sss.pgh.pa.us>	2004-06-18 06:14:31 +0000
committer	Tom Lane <tgl@sss.pgh.pa.us>	2004-06-18 06:14:31 +0000
commit	2467394ee1566e82d0314d12a0d1c0a5670a28c9 (patch)
tree	57b87b8c181a9c3eb0f33bf775a5f31b9de8b890 /src
parent	474875f4438ea0d18f9f4170117bc407e6812515 (diff)
download	postgresql-2467394ee1566e82d0314d12a0d1c0a5670a28c9.tar.gz postgresql-2467394ee1566e82d0314d12a0d1c0a5670a28c9.zip