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.

7 files changed, 375 insertions, 39 deletions

diff --git a/src/backend/catalog/Makefile b/src/backend/catalog/Makefile
index 0a68cb661a..c63168a9bd 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 de74a422b7..6c966b89b2 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 afea81a976..32d4d33d89 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 a4cc33d4c9..68ea6f45f6 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 8ada0915bd..581799fc5f 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 c7535508a7..b412023fe2 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 c3546f9068..c600ac2a84 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);