Index: src/backend/access/transam/xlog.c
===================================================================
RCS file: /projects/cvsroot/pgsql/src/backend/access/transam/xlog.c,v
retrieving revision 1.181
diff -c -r1.181 xlog.c
*** src/backend/access/transam/xlog.c	12 Feb 2005 23:53:37 -0000	1.181
--- src/backend/access/transam/xlog.c	5 Mar 2005 22:33:56 -0000
***************
*** 43,48 ****
--- 43,55 ----
  #include "utils/guc.h"
  #include "utils/relcache.h"
  
+ #include "catalog/pg_tablespace.h"
+ #include "catalog/catalog.h"
+ #include "access/skey.h"
+ #include "utils/fmgroids.h"
+ #include "access/relscan.h"
+ #include "access/heapam.h"
+ #include "utils/resowner.h"
  
  /*
   * This chunk of hackery attempts to determine which file sync methods
***************
*** 465,470 ****
--- 472,480 ----
  static bool read_backup_label(XLogRecPtr *checkPointLoc);
  static void remove_backup_label(void);
  
+ static void CleanupStaleRelFilesFrom(Oid tablespaceoid, Oid dboid);
+ static void CleanupStaleRelFilesFromTablespace(Oid tablespaceoid);
+ static void CleanupStaleRelFiles(void);
  
  /*
   * Insert an XLOG record having the specified RMID and info bytes,
***************
*** 4483,4488 ****
--- 4493,4500 ----
  
  		CreateCheckPoint(true, true);
  
+ 		CleanupStaleRelFiles();
+ 
  		/*
  		 * Close down recovery environment
  		 */
***************
*** 5656,5658 ****
--- 5668,5852 ----
  					 errmsg("could not remove file \"%s\": %m",
  							labelfilepath)));
  }
+ 
+ /* Like AllocateDir, but ereports on failure */
+ static DIR *
+ AllocateDirChecked(char *path)
+ {
+ 	DIR *dirdesc = AllocateDir(path);
+ 	if (dirdesc == NULL)
+ 		ereport(ERROR,
+ 				(errcode_for_file_access(),
+ 				 errmsg("could not open directory \"%s\": %m", 
+ 						path)));
+ 
+ 	return dirdesc;
+ }
+ 
+ /*
+  * Scan through all tablespaces for relations left over
+  * by aborted transactions. 
+  *
+  * For example, if a transaction issues
+  * BEGIN; CREATE TABLE foobar ();
+  * and then the backend crashes, the file is left in the
+  * tablespace until CleanupStaleRelFiles deletes it.
+  */
+ static void
+ CleanupStaleRelFiles(void)
+ {
+ 	DIR                *dirdesc;
+ 	struct dirent *de;
+ 	char	   *path;
+ 	int pathlen;
+ 
+ 	pathlen = strlen(DataDir) + 11 + 1;
+ 	path = (char *) palloc(pathlen);
+ 	snprintf(path, pathlen, "%s/pg_tblspc/", DataDir);
+ 
+ 	dirdesc = AllocateDirChecked(path);
+ 	
+ 	while ((de = readdir(dirdesc)) != NULL)
+ 	{
+ 		char *invalid;
+ 		Oid tablespaceoid;
+ 
+ 		tablespaceoid = (Oid) strtol(de->d_name, &invalid, 10);
+ 		if(invalid[0] == '\0')
+ 			CleanupStaleRelFilesFromTablespace(tablespaceoid);
+ 	}
+ 	pfree(path);
+ 
+ 	CleanupStaleRelFilesFromTablespace(DEFAULTTABLESPACE_OID);
+ }
+ 
+ /* Scan a specific tablespace for stale relations */
+ static void
+ CleanupStaleRelFilesFromTablespace(Oid tablespaceoid)
+ {
+ 	DIR                *dirdesc;
+ 	struct dirent *de;
+ 	char	   *path;
+ 
+ 	path = GetTablespacePath(tablespaceoid);
+ 
+ 	dirdesc = AllocateDirChecked(path);
+ 	while ((de = readdir(dirdesc)) != NULL)
+ 	{
+ 		char *invalid;
+ 		Oid dboid;
+ 
+ 		dboid = (Oid) strtol(de->d_name, &invalid, 10);
+ 		if(invalid[0] == '\0')
+ 			CleanupStaleRelFilesFrom(tablespaceoid, dboid);
+ 	}
+ 	pfree(path);
+ }
+ 
+ /* Scan a specific database in a specific tablespace for stale relations.
+  *
+  * First, pg_class for the database is opened, and the relfilenodes of all
+  * relations mentioned there are stored in a hash table.
+  *
+  * Then the directory is scanned. Every file in the directory that's not
+  * found in pg_class (the hash table) is deleted.
+  */
+ static void
+ CleanupStaleRelFilesFrom(Oid tablespaceoid, Oid dboid)
+ {
+ 	DIR *dirdesc;
+ 	struct dirent *de;
+ 	HASHCTL hashctl;
+ 	HTAB *relfilenodeHash;
+ 	MemoryContext mcxt;
+ 	RelFileNode rnode;
+ 	char *path;
+ 
+ 	/* We create a private memory context so that we can easily deallocate 
+ 	 * the hash table and its contents
+ 	 */
+ 	mcxt = AllocSetContextCreate(TopMemoryContext, "CleanupStaleRelFiles", 
+ 								 ALLOCSET_DEFAULT_MINSIZE,
+ 								 ALLOCSET_DEFAULT_INITSIZE, 
+ 								 ALLOCSET_DEFAULT_MAXSIZE);
+ 	
+ 	hashctl.hash = tag_hash;
+ 	/* The entry contents is not used for anything, we just check
+ 	 * if an oid is in the hash table or not. */
+ 	hashctl.keysize = sizeof(Oid);
+ 	hashctl.entrysize = 1;
+ 	hashctl.hcxt = mcxt;
+ 	relfilenodeHash = hash_create("relfilenodeHash", 100, &hashctl,
+ 								  HASH_FUNCTION | HASH_ELEM | HASH_CONTEXT);
+ 
+ 	/* Read all relfilenodes from pg_class into the hash table */
+ 	{
+ 		ResourceOwner owner;
+ 		ResourceOwner oldowner;
+ 		Relation	rel;
+ 		HeapScanDesc scan;
+ 		HeapTuple tuple;
+ 
+ 		/* Need a resowner to keep the heapam and buffer code happy */
+ 		owner = ResourceOwnerCreate(NULL, "CleanupStaleRelFiles");
+ 		oldowner = CurrentResourceOwner;
+ 		CurrentResourceOwner = owner;
+ 
+ 		rnode.spcNode = tablespaceoid;
+ 		rnode.dbNode = dboid;
+ 		rnode.relNode = RelOid_pg_class;
+ 		rel = XLogOpenRelation(true, 0 , rnode);
+ 
+ 		scan = heap_beginscan(rel, SnapshotNow, 0, NULL);
+ 		while ((tuple = heap_getnext(scan, ForwardScanDirection)) != NULL)
+ 		{
+ 			Form_pg_class classform = (Form_pg_class) GETSTRUCT(tuple);
+ 
+ 			hash_search(relfilenodeHash, &classform->relfilenode, 
+ 						HASH_ENTER, NULL);
+ 		}
+ 		heap_endscan(scan);
+ 
+ 		XLogCloseRelation(rnode);
+ 
+ 		CurrentResourceOwner = oldowner;  
+ 		ResourceOwnerDelete(owner);
+ 	}
+ 
+ 	/* Scan the directory */
+ 	path = GetDatabasePath(dboid, tablespaceoid);
+ 
+ 	dirdesc = AllocateDirChecked(path);
+ 	while ((de = readdir(dirdesc)) != NULL)
+ 	{
+ 		char *invalid;
+ 		Oid relfilenode;
+ 
+ 		relfilenode = strtol(de->d_name, &invalid, 10);
+ 		if(invalid[0] == '\0') 
+ 		{
+ 			/* Filename was a valid number, check if pg_class knows about it */
+ 			if(hash_search(relfilenodeHash, &relfilenode, 
+ 						   HASH_FIND, NULL) == NULL) 
+ 			{
+ 				char *filepath;
+ 				rnode.spcNode = tablespaceoid;
+ 				rnode.dbNode = dboid;
+ 				rnode.relNode = relfilenode;
+ 
+ 				filepath = relpath(rnode);
+ 
+ 				/*	unlink(filepath); */
+ 				ereport(WARNING,
+ 						(errcode_for_file_access(),
+ 						 errmsg("The file \"%s\" can be safely deleted",
+ 								filepath)));
+ 				pfree(filepath);
+ 			}
+ 		}
+ 	}
+ 	pfree(path);
+ 	hash_destroy(relfilenodeHash);
+ 	MemoryContextDelete(mcxt);
+ }
+ 
Index: src/backend/catalog/catalog.c
===================================================================
RCS file: /projects/cvsroot/pgsql/src/backend/catalog/catalog.c,v
retrieving revision 1.57
diff -c -r1.57 catalog.c
*** src/backend/catalog/catalog.c	31 Dec 2004 21:59:38 -0000	1.57
--- src/backend/catalog/catalog.c	5 Mar 2005 22:33:56 -0000
***************
*** 107,112 ****
--- 107,145 ----
  	return path;
  }
  
+ /*
+  * GetTablespacePath	- construct path to a tablespace symbolic link
+  *
+  * Result is a palloc'd string.
+  *
+  * XXX this must agree with relpath and GetDatabasePath!
+  */
+ char *
+ GetTablespacePath(Oid spcNode)
+ {
+ 	int			pathlen;
+ 	char	   *path;
+ 
+ 	Assert(spcNode != GLOBALTABLESPACE_OID);
+ 
+ 	if (spcNode == DEFAULTTABLESPACE_OID)
+ 	{
+ 		/* The default tablespace is {datadir}/base */
+ 		pathlen = strlen(DataDir) + 5 + 1;
+ 		path = (char *) palloc(pathlen);
+ 		snprintf(path, pathlen, "%s/base",
+ 				 DataDir);
+ 	}
+ 	else
+ 	{
+ 		/* All other tablespaces have symlinks in pg_tblspc */
+ 		pathlen = strlen(DataDir) + 11 + OIDCHARS + 1;
+ 		path = (char *) palloc(pathlen);
+ 		snprintf(path, pathlen, "%s/pg_tblspc/%u",
+ 				 DataDir, spcNode);
+ 	}
+ 	return path;
+ }
  
  /*
   * IsSystemRelation
Index: src/backend/commands/tablespace.c
===================================================================
RCS file: /projects/cvsroot/pgsql/src/backend/commands/tablespace.c,v
retrieving revision 1.16
diff -c -r1.16 tablespace.c
*** src/backend/commands/tablespace.c	27 Jan 2005 23:23:55 -0000	1.16
--- src/backend/commands/tablespace.c	5 Mar 2005 22:33:56 -0000
***************
*** 342,349 ****
  	/*
  	 * All seems well, create the symlink
  	 */
! 	linkloc = (char *) palloc(strlen(DataDir) + 11 + 10 + 1);
! 	sprintf(linkloc, "%s/pg_tblspc/%u", DataDir, tablespaceoid);
  
  	if (symlink(location, linkloc) < 0)
  		ereport(ERROR,
--- 342,348 ----
  	/*
  	 * All seems well, create the symlink
  	 */
! 	linkloc = GetTablespacePath(tablespaceoid);
  
  	if (symlink(location, linkloc) < 0)
  		ereport(ERROR,
***************
*** 496,503 ****
  	char	   *subfile;
  	struct stat st;
  
! 	location = (char *) palloc(strlen(DataDir) + 11 + 10 + 1);
! 	sprintf(location, "%s/pg_tblspc/%u", DataDir, tablespaceoid);
  
  	/*
  	 * Check if the tablespace still contains any files.  We try to rmdir
--- 495,501 ----
  	char	   *subfile;
  	struct stat st;
  
! 	location = GetTablespacePath(tablespaceoid);
  
  	/*
  	 * Check if the tablespace still contains any files.  We try to rmdir
***************
*** 1037,1044 ****
  		set_short_version(location);
  
  		/* Create the symlink if not already present */
! 		linkloc = (char *) palloc(strlen(DataDir) + 11 + 10 + 1);
! 		sprintf(linkloc, "%s/pg_tblspc/%u", DataDir, xlrec->ts_id);
  
  		if (symlink(location, linkloc) < 0)
  		{
--- 1035,1041 ----
  		set_short_version(location);
  
  		/* Create the symlink if not already present */
! 		linkloc = GetTablespacePath(xlrec->ts_id);
  
  		if (symlink(location, linkloc) < 0)
  		{
Index: src/include/catalog/catalog.h
===================================================================
RCS file: /projects/cvsroot/pgsql/src/include/catalog/catalog.h,v
retrieving revision 1.30
diff -c -r1.30 catalog.h
*** src/include/catalog/catalog.h	31 Dec 2004 22:03:24 -0000	1.30
--- src/include/catalog/catalog.h	5 Mar 2005 22:33:57 -0000
***************
*** 19,24 ****
--- 19,25 ----
  
  extern char *relpath(RelFileNode rnode);
  extern char *GetDatabasePath(Oid dbNode, Oid spcNode);
+ extern char *GetTablespacePath(Oid spcNode);
  
  extern bool IsSystemRelation(Relation relation);
  extern bool IsToastRelation(Relation relation);