Index: src/backend/catalog/catalog.c
===================================================================
RCS file: /cvsroot/pgsql/src/backend/catalog/catalog.c,v
retrieving revision 1.86
diff -c -c -r1.86 catalog.c
*** src/backend/catalog/catalog.c	6 Jan 2010 02:41:37 -0000	1.86
--- src/backend/catalog/catalog.c	8 Jan 2010 01:07:41 -0000
***************
*** 115,130 ****
  	else
  	{
  		/* All other tablespaces are accessed via symlinks */
! 		pathlen = 10 + OIDCHARS + 1 + OIDCHARS + 1 + OIDCHARS + 1
! 			+ FORKNAMECHARS + 1;
  		path = (char *) palloc(pathlen);
  		if (forknum != MAIN_FORKNUM)
! 			snprintf(path, pathlen, "pg_tblspc/%u/%u/%u_%s",
! 					 rnode.spcNode, rnode.dbNode, rnode.relNode,
! 					 forkNames[forknum]);
  		else
! 			snprintf(path, pathlen, "pg_tblspc/%u/%u/%u",
! 					 rnode.spcNode, rnode.dbNode, rnode.relNode);
  	}
  	return path;
  }
--- 115,131 ----
  	else
  	{
  		/* All other tablespaces are accessed via symlinks */
! 		pathlen = 9 + 1 + OIDCHARS + 1 + strlen(TABLESPACE_VERSION_DIRECTORY) +
! 			1 + OIDCHARS + 1 + OIDCHARS + 1 + FORKNAMECHARS + 1;
  		path = (char *) palloc(pathlen);
  		if (forknum != MAIN_FORKNUM)
! 			snprintf(path, pathlen, "pg_tblspc/%u/%s/%u/%u_%s",
! 					 rnode.spcNode, TABLESPACE_VERSION_DIRECTORY,
! 					 rnode.dbNode, rnode.relNode, forkNames[forknum]);
  		else
! 			snprintf(path, pathlen, "pg_tblspc/%u/%s/%u/%u",
! 					 rnode.spcNode, TABLESPACE_VERSION_DIRECTORY,
! 					 rnode.dbNode, rnode.relNode);
  	}
  	return path;
  }
***************
*** 161,170 ****
  	else
  	{
  		/* All other tablespaces are accessed via symlinks */
! 		pathlen = 10 + OIDCHARS + 1 + OIDCHARS + 1;
  		path = (char *) palloc(pathlen);
! 		snprintf(path, pathlen, "pg_tblspc/%u/%u",
! 				 spcNode, dbNode);
  	}
  	return path;
  }
--- 162,172 ----
  	else
  	{
  		/* All other tablespaces are accessed via symlinks */
! 		pathlen = 9 + 1 + OIDCHARS + 1 + strlen(TABLESPACE_VERSION_DIRECTORY) +
! 				  1 + OIDCHARS + 1;
  		path = (char *) palloc(pathlen);
! 		snprintf(path, pathlen, "pg_tblspc/%u/%s/%u",
! 				 spcNode, TABLESPACE_VERSION_DIRECTORY, dbNode);
  	}
  	return path;
  }
Index: src/backend/commands/tablespace.c
===================================================================
RCS file: /cvsroot/pgsql/src/backend/commands/tablespace.c,v
retrieving revision 1.70
diff -c -c -r1.70 tablespace.c
*** src/backend/commands/tablespace.c	7 Jan 2010 04:10:39 -0000	1.70
--- src/backend/commands/tablespace.c	8 Jan 2010 01:07:41 -0000
***************
*** 15,22 ****
   * To support file access via the information given in RelFileNode, we
   * maintain a symbolic-link map in $PGDATA/pg_tblspc. 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_tblspc/spcoid/dboid/relfilenode
   *
   * There are two tablespaces created at initdb time: pg_global (for shared
   * tables) and pg_default (for everything else).  For backwards compatibility
--- 15,25 ----
   * To support file access via the information given in RelFileNode, we
   * maintain a symbolic-link map in $PGDATA/pg_tblspc. The symlinks are
   * named by tablespace OIDs and point to the actual tablespace directories.
+  * There is also a per-cluster version directory in each tablespace.
   * Thus the full path to an arbitrary file is
!  *			$PGDATA/pg_tblspc/spcoid/PG_MAJORVER_CATVER/dboid/relfilenode
!  * e.g.
!  *			$PGDATA/pg_tblspc/20981/PG_8.5_201001061/719849/83292814
   *
   * There are two tablespaces created at initdb time: pg_global (for shared
   * tables) and pg_default (for everything else).  For backwards compatibility
***************
*** 81,88 ****
  char	   *temp_tablespaces = NULL;
  
  
! static bool remove_tablespace_directories(Oid tablespaceoid, bool redo);
! static void write_version_file(const char *path);
  
  
  /*
--- 84,92 ----
  char	   *temp_tablespaces = NULL;
  
  
! static void create_tablespace_directories(const char *location,
! 										 const Oid tablespaceoid);
! static bool destroy_tablespace_directories(Oid tablespaceoid, bool redo);
  
  
  /*
***************
*** 146,163 ****
  				{
  					char	   *parentdir;
  
! 					/* Failure other than not exists? */
  					if (errno != ENOENT || !isRedo)
  						ereport(ERROR,
  								(errcode_for_file_access(),
  							  errmsg("could not create directory \"%s\": %m",
  									 dir)));
  
! 					/* Parent directory must be missing */
  					parentdir = pstrdup(dir);
  					get_parent_directory(parentdir);
! 					/* Can't create parent either? */
! 					if (mkdir(parentdir, S_IRWXU) < 0)
  						ereport(ERROR,
  								(errcode_for_file_access(),
  							  errmsg("could not create directory \"%s\": %m",
--- 150,185 ----
  				{
  					char	   *parentdir;
  
! 					/* Failure other than not exists or not in WAL replay? */
  					if (errno != ENOENT || !isRedo)
  						ereport(ERROR,
  								(errcode_for_file_access(),
  							  errmsg("could not create directory \"%s\": %m",
  									 dir)));
  
! 					/*
! 					 * Parent directories are missing during WAL replay, so
! 					 * continue by creating simple parent directories
! 					 * rather than a symlink.
! 					 */
! 
! 					/* create two parents up if not exist */
  					parentdir = pstrdup(dir);
  					get_parent_directory(parentdir);
! 					get_parent_directory(parentdir);
! 					/* Can't create parent and it doesn't already exist? */
! 					if (mkdir(parentdir, S_IRWXU) < 0 && errno != EEXIST)
! 						ereport(ERROR,
! 								(errcode_for_file_access(),
! 							  errmsg("could not create directory \"%s\": %m",
! 									 parentdir)));
! 					pfree(parentdir);
! 
! 					/* create one parent up if not exist */
! 					parentdir = pstrdup(dir);
! 					get_parent_directory(parentdir);
! 					/* Can't create parent and it doesn't already exist? */
! 					if (mkdir(parentdir, S_IRWXU) < 0 && errno != EEXIST)
  						ereport(ERROR,
  								(errcode_for_file_access(),
  							  errmsg("could not create directory \"%s\": %m",
***************
*** 212,218 ****
  	HeapTuple	tuple;
  	Oid			tablespaceoid;
  	char	   *location;
- 	char	   *linkloc;
  	Oid			ownerId;
  
  	/* Must be super user */
--- 234,239 ----
***************
*** 251,260 ****
  
  	/*
  	 * 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 - OIDCHARS - 1 - OIDCHARS - 1 - OIDCHARS)
  		ereport(ERROR,
  				(errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
  				 errmsg("tablespace location \"%s\" is too long",
--- 272,282 ----
  
  	/*
  	 * Check that location isn't too long. Remember that we're going to append
! 	 * 'PG_XXX/<dboid>/<relid>.<nnn>'.  FYI, we never actually reference the
! 	 * whole path, but mkdir() uses the first two parts.
  	 */
! 	if (strlen(location) + 1 + strlen(TABLESPACE_VERSION_DIRECTORY) + 1 +
! 		OIDCHARS + 1 + OIDCHARS + 1 + OIDCHARS > MAXPGPATH)
  		ereport(ERROR,
  				(errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
  				 errmsg("tablespace location \"%s\" is too long",
***************
*** 311,355 ****
  	/* Record dependency on owner */
  	recordDependencyOnOwner(TableSpaceRelationId, tablespaceoid, ownerId);
  
! 	/*
! 	 * 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.
! 	 */
! 	write_version_file(location);
! 
! 	/*
! 	 * All seems well, create the symlink
! 	 */
! 	linkloc = (char *) palloc(OIDCHARS + OIDCHARS + 1);
! 	sprintf(linkloc, "pg_tblspc/%u", tablespaceoid);
! 
! 	if (symlink(location, linkloc) < 0)
! 		ereport(ERROR,
! 				(errcode_for_file_access(),
! 				 errmsg("could not create symbolic link \"%s\": %m",
! 						linkloc)));
  
  	/* Record the filesystem change in XLOG */
  	{
--- 333,339 ----
  	/* Record dependency on owner */
  	recordDependencyOnOwner(TableSpaceRelationId, tablespaceoid, ownerId);
  
! 	create_tablespace_directories(location, tablespaceoid);
  
  	/* Record the filesystem change in XLOG */
  	{
***************
*** 378,384 ****
  	 */
  	ForceSyncCommit();
  
- 	pfree(linkloc);
  	pfree(location);
  
  	/* We keep the lock on pg_tablespace until commit */
--- 362,367 ----
***************
*** 478,484 ****
  	/*
  	 * Try to remove the physical infrastructure.
  	 */
! 	if (!remove_tablespace_directories(tablespaceoid, false))
  	{
  		/*
  		 * Not all files deleted?  However, there can be lingering empty files
--- 461,467 ----
  	/*
  	 * Try to remove the physical infrastructure.
  	 */
! 	if (!destroy_tablespace_directories(tablespaceoid, false))
  	{
  		/*
  		 * Not all files deleted?  However, there can be lingering empty files
***************
*** 490,496 ****
  		 * out any lingering files, and try again.
  		 */
  		RequestCheckpoint(CHECKPOINT_IMMEDIATE | CHECKPOINT_FORCE | CHECKPOINT_WAIT);
! 		if (!remove_tablespace_directories(tablespaceoid, false))
  		{
  			/* Still not empty, the files must be important then */
  			ereport(ERROR,
--- 473,479 ----
  		 * out any lingering files, and try again.
  		 */
  		RequestCheckpoint(CHECKPOINT_IMMEDIATE | CHECKPOINT_FORCE | CHECKPOINT_WAIT);
! 		if (!destroy_tablespace_directories(tablespaceoid, false))
  		{
  			/* Still not empty, the files must be important then */
  			ereport(ERROR,
***************
*** 542,565 ****
  #endif   /* HAVE_SYMLINK */
  }
  
  /*
!  * remove_tablespace_directories: attempt to remove filesystem infrastructure
   *
!  * Returns TRUE if successful, FALSE if some subdirectory is not empty
   *
!  * redo indicates we are redoing a drop from XLOG; okay if nothing there
   */
  static bool
! remove_tablespace_directories(Oid tablespaceoid, bool redo)
  {
! 	char	   *location;
  	DIR		   *dirdesc;
  	struct dirent *de;
  	char	   *subfile;
  	struct stat st;
  
! 	location = (char *) palloc(OIDCHARS + OIDCHARS + 1);
! 	sprintf(location, "pg_tblspc/%u", tablespaceoid);
  
  	/*
  	 * Check if the tablespace still contains any files.  We try to rmdir each
--- 525,621 ----
  #endif   /* HAVE_SYMLINK */
  }
  
+ 
  /*
!  * create_tablespace_directories
   *
!  *	Attempt to create filesystem infrastructure linking $PGDATA/pg_tblspc/
!  *	to the specified directory
!  */
! static void
! create_tablespace_directories(const char *location, const Oid tablespaceoid)
! {
! 	char *linkloc = palloc(OIDCHARS + OIDCHARS + 1);
! 	char *location_with_version_dir = palloc(strlen(location) + 1 +
! 										strlen(TABLESPACE_VERSION_DIRECTORY) + 1);
! 
! 	sprintf(linkloc, "pg_tblspc/%u", tablespaceoid);
! 	sprintf(location_with_version_dir, "%s/%s", location,
! 										TABLESPACE_VERSION_DIRECTORY);
! 
! 	/*
! 	 * 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)
! 	{
! 		if (errno == ENOENT)
! 			ereport(ERROR,
! 					(errcode(ERRCODE_UNDEFINED_FILE),
! 					 errmsg("directory \"%s\" does not exist",
! 							location)));
! 		else
! 			ereport(ERROR,
! 				(errcode_for_file_access(),
! 				 errmsg("could not set permissions on directory \"%s\": %m",
! 						location)));
! 	}
! 
! 	/*
! 	 * The creation of the version directory prevents more than one
! 	 * 	tablespace in a single location.
! 	 */
! 	if (mkdir(location_with_version_dir, S_IRWXU) < 0)
! 	{
! 		if (errno == EEXIST)
! 			ereport(ERROR,
! 					(errcode(ERRCODE_OBJECT_IN_USE),
! 					 errmsg("directory \"%s\" already in use as a tablespace",
! 							location_with_version_dir)));
! 		else
! 			ereport(ERROR,
! 					(errcode_for_file_access(),
! 				  errmsg("could not create directory \"%s\": %m",
! 						 location_with_version_dir)));
! 	}
! 
! 	/*
! 	 * Create the symlink under PGDATA
! 	 */
! 	if (symlink(location, linkloc) < 0)
! 		ereport(ERROR,
! 				(errcode_for_file_access(),
! 				 errmsg("could not create symbolic link \"%s\": %m",
! 						linkloc)));
! 
! 	pfree(linkloc);
! 	pfree(location_with_version_dir);
! }
! 
! 
! /*
!  * destroy_tablespace_directories
!  *
!  * Attempt to remove filesystem infrastructure
!  *
!  * 'redo' indicates we are redoing a drop from XLOG; okay if nothing there
   *
!  * Returns TRUE if successful, FALSE if some subdirectory is not empty
   */
  static bool
! destroy_tablespace_directories(Oid tablespaceoid, bool redo)
  {
! 	char	   *linkloc;
! 	char	   *linkloc_with_version_dir;
  	DIR		   *dirdesc;
  	struct dirent *de;
  	char	   *subfile;
  	struct stat st;
  
! 	linkloc_with_version_dir = palloc(9 + 1 + OIDCHARS + 1 +
! 									strlen(TABLESPACE_VERSION_DIRECTORY));
! 	sprintf(linkloc_with_version_dir, "pg_tblspc/%u/%s", tablespaceoid,
! 									TABLESPACE_VERSION_DIRECTORY);
  
  	/*
  	 * Check if the tablespace still contains any files.  We try to rmdir each
***************
*** 582,588 ****
  	 * and symlink.  We want to allow a new DROP attempt to succeed at
  	 * removing the catalog entries, so we should not give a hard error here.
  	 */
! 	dirdesc = AllocateDir(location);
  	if (dirdesc == NULL)
  	{
  		if (errno == ENOENT)
--- 638,644 ----
  	 * and symlink.  We want to allow a new DROP attempt to succeed at
  	 * removing the catalog entries, so we should not give a hard error here.
  	 */
! 	dirdesc = AllocateDir(linkloc_with_version_dir);
  	if (dirdesc == NULL)
  	{
  		if (errno == ENOENT)
***************
*** 591,622 ****
  				ereport(WARNING,
  						(errcode_for_file_access(),
  						 errmsg("could not open directory \"%s\": %m",
! 								location)));
! 			pfree(location);
  			return true;
  		}
  		/* else let ReadDir report the error */
  	}
  
! 	while ((de = ReadDir(dirdesc, location)) != 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)
  			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))
  		{
  			FreeDir(dirdesc);
  			return false;
  		}
  
! 		/* Do the real deed */
  		if (rmdir(subfile) < 0)
  			ereport(ERROR,
  					(errcode_for_file_access(),
--- 647,678 ----
  				ereport(WARNING,
  						(errcode_for_file_access(),
  						 errmsg("could not open directory \"%s\": %m",
! 								linkloc_with_version_dir)));
! 			pfree(linkloc_with_version_dir);
  			return true;
  		}
  		/* else let ReadDir report the error */
  	}
  
! 	while ((de = ReadDir(dirdesc, linkloc_with_version_dir)) != NULL)
  	{
  		if (strcmp(de->d_name, ".") == 0 ||
! 			strcmp(de->d_name, "..") == 0)
  			continue;
  
! 		subfile = palloc(strlen(linkloc_with_version_dir) + 1 + strlen(de->d_name) + 1);
! 		sprintf(subfile, "%s/%s", linkloc_with_version_dir, de->d_name);
  
  		/* This check is just to deliver a friendlier error message */
  		if (!directory_is_empty(subfile))
  		{
  			FreeDir(dirdesc);
+ 			pfree(subfile);
+ 			pfree(linkloc_with_version_dir);
  			return false;
  		}
  
! 		/* remove empty directory */
  		if (rmdir(subfile) < 0)
  			ereport(ERROR,
  					(errcode_for_file_access(),
***************
*** 628,706 ****
  
  	FreeDir(dirdesc);
  
  	/*
! 	 * Okay, try to unlink PG_VERSION (we allow it to not be there, even in
! 	 * non-REDO case, for robustness).
! 	 */
! 	subfile = palloc(strlen(location) + 11 + 1);
! 	sprintf(subfile, "%s/PG_VERSION", location);
! 
! 	if (unlink(subfile) < 0)
! 	{
! 		if (errno != ENOENT)
! 			ereport(ERROR,
! 					(errcode_for_file_access(),
! 					 errmsg("could not remove file \"%s\": %m",
! 							subfile)));
! 	}
! 
! 	pfree(subfile);
! 
! 	/*
! 	 * Okay, try to remove the symlink.  We must however deal with the
  	 * possibility that it's a directory instead of a symlink --- this could
  	 * happen during WAL replay (see TablespaceCreateDbspace), and it is also
! 	 * the normal case on Windows.
  	 */
! 	if (lstat(location, &st) == 0 && S_ISDIR(st.st_mode))
  	{
! 		if (rmdir(location) < 0)
  			ereport(ERROR,
  					(errcode_for_file_access(),
  					 errmsg("could not remove directory \"%s\": %m",
! 							location)));
  	}
  	else
  	{
! 		if (unlink(location) < 0)
  			ereport(ERROR,
  					(errcode_for_file_access(),
  					 errmsg("could not remove symbolic link \"%s\": %m",
! 							location)));
  	}
  
! 	pfree(location);
  
  	return true;
  }
  
- /*
-  * write out the PG_VERSION file in the specified directory
-  */
- static void
- write_version_file(const char *path)
- {
- 	char	   *fullname;
- 	FILE	   *version_file;
- 
- 	/* Now write the file */
- 	fullname = palloc(strlen(path) + 11 + 1);
- 	sprintf(fullname, "%s/PG_VERSION", path);
- 
- 	if ((version_file = AllocateFile(fullname, PG_BINARY_W)) == NULL)
- 		ereport(ERROR,
- 				(errcode_for_file_access(),
- 				 errmsg("could not write to file \"%s\": %m",
- 						fullname)));
- 	fprintf(version_file, "%s\n", PG_MAJORVERSION);
- 	if (FreeFile(version_file))
- 		ereport(ERROR,
- 				(errcode_for_file_access(),
- 				 errmsg("could not write to file \"%s\": %m",
- 						fullname)));
- 
- 	pfree(fullname);
- }
  
  /*
   * Check if a directory is empty.
--- 684,727 ----
  
  	FreeDir(dirdesc);
  
+ 	/* remove version directory */
+ 	if (rmdir(linkloc_with_version_dir) < 0)
+ 		ereport(ERROR,
+ 				(errcode_for_file_access(),
+ 				 errmsg("could not remove directory \"%s\": %m",
+ 						linkloc_with_version_dir)));
+  
  	/*
! 	 * Try to remove the symlink.  We must however deal with the
  	 * possibility that it's a directory instead of a symlink --- this could
  	 * happen during WAL replay (see TablespaceCreateDbspace), and it is also
! 	 * the case on Windows where junction points lstat() as directories.
  	 */
! 	linkloc = pstrdup(linkloc_with_version_dir);
! 	get_parent_directory(linkloc);
! 	if (lstat(linkloc, &st) == 0 && S_ISDIR(st.st_mode))
  	{
! 		if (rmdir(linkloc) < 0)
  			ereport(ERROR,
  					(errcode_for_file_access(),
  					 errmsg("could not remove directory \"%s\": %m",
! 							linkloc)));
  	}
  	else
  	{
! 		if (unlink(linkloc) < 0)
  			ereport(ERROR,
  					(errcode_for_file_access(),
  					 errmsg("could not remove symbolic link \"%s\": %m",
! 							linkloc)));
  	}
  
! 	pfree(linkloc_with_version_dir);
! 	pfree(linkloc);
  
  	return true;
  }
  
  
  /*
   * Check if a directory is empty.
***************
*** 728,733 ****
--- 749,755 ----
  	return true;
  }
  
+ 
  /*
   * Rename a tablespace
   */
***************
*** 1336,1370 ****
  	{
  		xl_tblspc_create_rec *xlrec = (xl_tblspc_create_rec *) XLogRecGetData(record);
  		char	   *location = xlrec->ts_path;
- 		char	   *linkloc;
- 
- 		/*
- 		 * 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)));
- 
- 		/* Create or re-create the PG_VERSION file in the target directory */
- 		write_version_file(location);
- 
- 		/* Create the symlink if not already present */
- 		linkloc = (char *) palloc(OIDCHARS + OIDCHARS + 1);
- 		sprintf(linkloc, "pg_tblspc/%u", xlrec->ts_id);
- 
- 		if (symlink(location, linkloc) < 0)
- 		{
- 			if (errno != EEXIST)
- 				ereport(ERROR,
- 						(errcode_for_file_access(),
- 						 errmsg("could not create symbolic link \"%s\": %m",
- 								linkloc)));
- 		}
  
! 		pfree(linkloc);
  	}
  	else if (info == XLOG_TBLSPC_DROP)
  	{
--- 1358,1365 ----
  	{
  		xl_tblspc_create_rec *xlrec = (xl_tblspc_create_rec *) XLogRecGetData(record);
  		char	   *location = xlrec->ts_path;
  
! 		create_tablespace_directories(location, xlrec->ts_id);
  	}
  	else if (info == XLOG_TBLSPC_DROP)
  	{
***************
*** 1380,1386 ****
  		 * remove all files then do conflict processing and try again,
  		 * if currently enabled.
  		 */
! 		if (!remove_tablespace_directories(xlrec->ts_id, true))
  		{
  			VirtualTransactionId *temp_file_users;
  
--- 1375,1381 ----
  		 * remove all files then do conflict processing and try again,
  		 * if currently enabled.
  		 */
! 		if (!destroy_tablespace_directories(xlrec->ts_id, true))
  		{
  			VirtualTransactionId *temp_file_users;
  
***************
*** 1416,1422 ****
  			 * exited by now. So lets recheck before we throw an error.
  			 * If !process_conflicts then this will just fail again.
  			 */
! 			if (!remove_tablespace_directories(xlrec->ts_id, true))
  				ereport(ERROR,
  					(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
  					 errmsg("tablespace %u is not empty",
--- 1411,1417 ----
  			 * exited by now. So lets recheck before we throw an error.
  			 * If !process_conflicts then this will just fail again.
  			 */
! 			if (!destroy_tablespace_directories(xlrec->ts_id, true))
  				ereport(ERROR,
  					(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
  					 errmsg("tablespace %u is not empty",
Index: src/backend/storage/file/fd.c
===================================================================
RCS file: /cvsroot/pgsql/src/backend/storage/file/fd.c,v
retrieving revision 1.152
diff -c -c -r1.152 fd.c
*** src/backend/storage/file/fd.c	2 Jan 2010 16:57:51 -0000	1.152
--- src/backend/storage/file/fd.c	8 Jan 2010 01:07:41 -0000
***************
*** 51,56 ****
--- 51,57 ----
  
  #include "miscadmin.h"
  #include "access/xact.h"
+ #include "catalog/catalog.h"
  #include "catalog/pg_tablespace.h"
  #include "storage/fd.h"
  #include "storage/ipc.h"
***************
*** 963,970 ****
  	else
  	{
  		/* All other tablespaces are accessed via symlinks */
! 		snprintf(tempdirpath, sizeof(tempdirpath), "pg_tblspc/%u/%s",
! 				 tblspcOid, PG_TEMP_FILES_DIR);
  	}
  
  	/*
--- 964,971 ----
  	else
  	{
  		/* All other tablespaces are accessed via symlinks */
! 		snprintf(tempdirpath, sizeof(tempdirpath), "pg_tblspc/%u/%s/%s",
! 				 tblspcOid, TABLESPACE_VERSION_DIRECTORY, PG_TEMP_FILES_DIR);
  	}
  
  	/*
***************
*** 1841,1848 ****
  			strcmp(spc_de->d_name, "..") == 0)
  			continue;
  
! 		snprintf(temp_path, sizeof(temp_path), "pg_tblspc/%s/%s",
! 				 spc_de->d_name, PG_TEMP_FILES_DIR);
  		RemovePgTempFilesInDir(temp_path);
  	}
  
--- 1842,1849 ----
  			strcmp(spc_de->d_name, "..") == 0)
  			continue;
  
! 		snprintf(temp_path, sizeof(temp_path), "pg_tblspc/%s/%s/%s",
! 				 spc_de->d_name, TABLESPACE_VERSION_DIRECTORY, PG_TEMP_FILES_DIR);
  		RemovePgTempFilesInDir(temp_path);
  	}
  
Index: src/backend/utils/adt/dbsize.c
===================================================================
RCS file: /cvsroot/pgsql/src/backend/utils/adt/dbsize.c,v
retrieving revision 1.25
diff -c -c -r1.25 dbsize.c
*** src/backend/utils/adt/dbsize.c	2 Jan 2010 16:57:53 -0000	1.25
--- src/backend/utils/adt/dbsize.c	8 Jan 2010 01:07:41 -0000
***************
*** 108,115 ****
  			strcmp(direntry->d_name, "..") == 0)
  			continue;
  
! 		snprintf(pathname, MAXPGPATH, "pg_tblspc/%s/%u",
! 				 direntry->d_name, dbOid);
  		totalsize += db_dir_size(pathname);
  	}
  
--- 108,115 ----
  			strcmp(direntry->d_name, "..") == 0)
  			continue;
  
! 		snprintf(pathname, MAXPGPATH, "pg_tblspc/%s/%s/%u",
! 				 direntry->d_name, TABLESPACE_VERSION_DIRECTORY, dbOid);
  		totalsize += db_dir_size(pathname);
  	}
  
***************
*** 179,185 ****
  	else if (tblspcOid == GLOBALTABLESPACE_OID)
  		snprintf(tblspcPath, MAXPGPATH, "global");
  	else
! 		snprintf(tblspcPath, MAXPGPATH, "pg_tblspc/%u", tblspcOid);
  
  	dirdesc = AllocateDir(tblspcPath);
  
--- 179,186 ----
  	else if (tblspcOid == GLOBALTABLESPACE_OID)
  		snprintf(tblspcPath, MAXPGPATH, "global");
  	else
! 		snprintf(tblspcPath, MAXPGPATH, "pg_tblspc/%u/%s", tblspcOid,
! 										 TABLESPACE_VERSION_DIRECTORY);
  
  	dirdesc = AllocateDir(tblspcPath);
  
Index: src/backend/utils/adt/misc.c
===================================================================
RCS file: /cvsroot/pgsql/src/backend/utils/adt/misc.c,v
retrieving revision 1.73
diff -c -c -r1.73 misc.c
*** src/backend/utils/adt/misc.c	2 Jan 2010 16:57:54 -0000	1.73
--- src/backend/utils/adt/misc.c	8 Jan 2010 01:07:41 -0000
***************
*** 20,25 ****
--- 20,26 ----
  #include <math.h>
  
  #include "access/xact.h"
+ #include "catalog/catalog.h"
  #include "catalog/pg_type.h"
  #include "catalog/pg_tablespace.h"
  #include "commands/dbcommands.h"
***************
*** 185,191 ****
  		/*
  		 * size = tablespace dirname length + dir sep char + oid + terminator
  		 */
! 		fctx->location = (char *) palloc(10 + 10 + 1);
  		if (tablespaceOid == GLOBALTABLESPACE_OID)
  		{
  			fctx->dirdesc = NULL;
--- 186,193 ----
  		/*
  		 * size = tablespace dirname length + dir sep char + oid + terminator
  		 */
! 		fctx->location = (char *) palloc(9 + 1 + OIDCHARS + 1 +
! 						 strlen(TABLESPACE_VERSION_DIRECTORY) + 1);
  		if (tablespaceOid == GLOBALTABLESPACE_OID)
  		{
  			fctx->dirdesc = NULL;
***************
*** 197,203 ****
  			if (tablespaceOid == DEFAULTTABLESPACE_OID)
  				sprintf(fctx->location, "base");
  			else
! 				sprintf(fctx->location, "pg_tblspc/%u", tablespaceOid);
  
  			fctx->dirdesc = AllocateDir(fctx->location);
  
--- 199,206 ----
  			if (tablespaceOid == DEFAULTTABLESPACE_OID)
  				sprintf(fctx->location, "base");
  			else
! 				sprintf(fctx->location, "pg_tblspc/%u/%s", tablespaceOid,
! 						TABLESPACE_VERSION_DIRECTORY);
  
  			fctx->dirdesc = AllocateDir(fctx->location);
  
Index: src/backend/utils/cache/relcache.c
===================================================================
RCS file: /cvsroot/pgsql/src/backend/utils/cache/relcache.c,v
retrieving revision 1.296
diff -c -c -r1.296 relcache.c
*** src/backend/utils/cache/relcache.c	5 Jan 2010 01:06:56 -0000	1.296
--- src/backend/utils/cache/relcache.c	8 Jan 2010 01:07:50 -0000
***************
*** 4297,4304 ****
  		if (strspn(de->d_name, "0123456789") == strlen(de->d_name))
  		{
  			/* Scan the tablespace dir for per-database dirs */
! 			snprintf(path, sizeof(path), "%s/%s",
! 					 tblspcdir, de->d_name);
  			RelationCacheInitFileRemoveInDir(path);
  		}
  	}
--- 4297,4304 ----
  		if (strspn(de->d_name, "0123456789") == strlen(de->d_name))
  		{
  			/* Scan the tablespace dir for per-database dirs */
! 			snprintf(path, sizeof(path), "%s/%s/%s",
! 					 tblspcdir, de->d_name, TABLESPACE_VERSION_DIRECTORY);
  			RelationCacheInitFileRemoveInDir(path);
  		}
  	}
Index: src/include/catalog/catalog.h
===================================================================
RCS file: /cvsroot/pgsql/src/include/catalog/catalog.h,v
retrieving revision 1.46
diff -c -c -r1.46 catalog.h
*** src/include/catalog/catalog.h	6 Jan 2010 02:41:37 -0000	1.46
--- src/include/catalog/catalog.h	8 Jan 2010 01:07:50 -0000
***************
*** 14,24 ****
--- 14,27 ----
  #ifndef CATALOG_H
  #define CATALOG_H
  
+ #include "catalog/catversion.h"
  #include "catalog/pg_class.h"
  #include "storage/relfilenode.h"
  #include "utils/relcache.h"
  
  #define OIDCHARS		10		/* max chars printed by %u */
+ #define TABLESPACE_VERSION_DIRECTORY	"PG_" PG_MAJORVERSION "_" \
+ 									CppAsString2(CATALOG_VERSION_NO)
  
  extern const char *forkNames[];
  extern ForkNumber forkname_to_number(char *forkName);
Index: src/test/regress/output/tablespace.source
===================================================================
RCS file: /cvsroot/pgsql/src/test/regress/output/tablespace.source,v
retrieving revision 1.9
diff -c -c -r1.9 tablespace.source
*** src/test/regress/output/tablespace.source	5 Jan 2010 21:54:00 -0000	1.9
--- src/test/regress/output/tablespace.source	8 Jan 2010 01:07:50 -0000
***************
*** 65,71 ****
  
  -- 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
--- 65,71 ----
  
  -- Will fail with bad path
  CREATE TABLESPACE badspace LOCATION '/no/such/location';
! ERROR:  directory "/no/such/location" does not exist
  -- No such tablespace
  CREATE TABLE bar (i int) TABLESPACE nosuchspace;
  ERROR:  tablespace "nosuchspace" does not exist