Add WAL logging for CREATE/DROP DATABASE and CREATE/DROP TABLESPACE.

Fix TablespaceCreateDbspace() to be able to create a dummy directory
in place of a dropped tablespace's symlink.  This eliminates the open
problem of a PANIC during WAL replay when a replayed action attempts
to touch a file in a since-deleted tablespace.  It also makes for a
significant improvement in the usability of PITR replay.

9 files changed, 483 insertions, 106 deletions

diff --git a/doc/src/sgml/backup.sgml b/doc/src/sgml/backup.sgml
index 7a3c4f73ff9..af06249203a 100644
--- a/doc/src/sgml/backup.sgml
+++ b/doc/src/sgml/backup.sgml
@@ -1,5 +1,5 @@
 <!--
-$PostgreSQL: pgsql/doc/src/sgml/backup.sgml,v 2.46 2004/08/08 04:34:43 tgl Exp $
+$PostgreSQL: pgsql/doc/src/sgml/backup.sgml,v 2.47 2004/08/29 21:08:47 tgl Exp $
 -->
 <chapter id="backup">
  <title>Backup and Restore</title>
@@ -902,19 +902,11 @@ restore_command = 'cp /mnt/server/archivedir/%f %p'
 
    <para>
     At this writing, there are several limitations of the on-line backup
-    technique.  These will probably be fixed in future releases.
+    technique.  These will probably be fixed in future releases:
 
   <itemizedlist>
    <listitem>
     <para>
-     The effects of <command>CREATE DATABASE</>, <command>DROP DATABASE</>,
-     <command>CREATE TABLESPACE</>, and <command>DROP TABLESPACE</> are
-     not fully reflected in the WAL log.  It is recommended that you take
-     a new base backup after performing one of these operations.
-    </para>
-   </listitem>
-   <listitem>
-    <para>
      Operations on non-btree indexes (hash, R-tree, and GiST indexes) are
      not presently WAL-logged, so replay will not update these index types.
      The recommended workaround, if you use any non-btree indexes, is to
@@ -932,7 +924,7 @@ restore_command = 'cp /mnt/server/archivedir/%f %p'
     since we may need to fix partially-written disk pages.  It is not
     necessary to store so many page copies for PITR operations, however.
     An area for future development is to compress archived WAL data by
-    removing unnecesssary page copies.
+    removing unnecessary page copies.
    </para>
   </sect2>
  </sect1>
diff --git a/src/backend/access/transam/rmgr.c b/src/backend/access/transam/rmgr.c
index 575ad7a0891..db51f36a009 100644
--- a/src/backend/access/transam/rmgr.c
+++ b/src/backend/access/transam/rmgr.c
@@ -3,20 +3,22 @@
  *
  * Resource managers definition
  *
- * $PostgreSQL: pgsql/src/backend/access/transam/rmgr.c,v 1.15 2004/08/23 23:22:44 tgl Exp $
+ * $PostgreSQL: pgsql/src/backend/access/transam/rmgr.c,v 1.16 2004/08/29 21:08:47 tgl Exp $
  */
 #include "postgres.h"
 
+#include "access/clog.h"
 #include "access/gist.h"
 #include "access/hash.h"
 #include "access/heapam.h"
 #include "access/nbtree.h"
 #include "access/rtree.h"
-#include "access/clog.h"
 #include "access/xact.h"
 #include "access/xlog_internal.h"
-#include "storage/smgr.h"
+#include "commands/dbcommands.h"
 #include "commands/sequence.h"
+#include "commands/tablespace.h"
+#include "storage/smgr.h"
 
 
 const RmgrData RmgrTable[RM_MAX_ID + 1] = {
@@ -24,8 +26,8 @@ const RmgrData RmgrTable[RM_MAX_ID + 1] = {
 	{"Transaction", xact_redo, xact_undo, xact_desc, NULL, NULL},
 	{"Storage", smgr_redo, smgr_undo, smgr_desc, NULL, NULL},
 	{"CLOG", clog_redo, clog_undo, clog_desc, NULL, NULL},
-	{"Reserved 4", NULL, NULL, NULL, NULL, NULL},
-	{"Reserved 5", NULL, NULL, NULL, NULL, NULL},
+	{"Database", dbase_redo, dbase_undo, dbase_desc, NULL, NULL},
+	{"Tablespace", tblspc_redo, tblspc_undo, tblspc_desc, NULL, NULL},
 	{"Reserved 6", NULL, NULL, NULL, NULL, NULL},
 	{"Reserved 7", NULL, NULL, NULL, NULL, NULL},
 	{"Reserved 8", NULL, NULL, NULL, NULL, NULL},
diff --git a/src/backend/commands/dbcommands.c b/src/backend/commands/dbcommands.c
index f7ef440b02e..aa95d97d0e2 100644
--- a/src/backend/commands/dbcommands.c
+++ b/src/backend/commands/dbcommands.c
@@ -9,7 +9,7 @@
  *
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/commands/dbcommands.c,v 1.141 2004/08/29 05:06:41 momjian Exp $
+ *	  $PostgreSQL: pgsql/src/backend/commands/dbcommands.c,v 1.142 2004/08/29 21:08:47 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -385,6 +385,30 @@ createdb(const CreatedbStmt *stmt)
 					(errmsg("could not initialize database directory")));
 		}
 #endif   /* WIN32 */
+
+		/* Record the filesystem change in XLOG */
+		{
+			xl_dbase_create_rec xlrec;
+			XLogRecData rdata[3];
+
+			xlrec.db_id = dboid;
+			rdata[0].buffer = InvalidBuffer;
+			rdata[0].data = (char *) &xlrec;
+			rdata[0].len = offsetof(xl_dbase_create_rec, src_path);
+			rdata[0].next = &(rdata[1]);
+
+			rdata[1].buffer = InvalidBuffer;
+			rdata[1].data = (char *) srcpath;
+			rdata[1].len = strlen(srcpath) + 1;
+			rdata[1].next = &(rdata[2]);
+
+			rdata[2].buffer = InvalidBuffer;
+			rdata[2].data = (char *) dstpath;
+			rdata[2].len = strlen(dstpath) + 1;
+			rdata[2].next = NULL;
+
+			(void) XLogInsert(RM_DBASE_ID, XLOG_DBASE_CREATE, rdata);
+		}
 	}
 	heap_endscan(scan);
 	heap_close(rel, AccessShareLock);
@@ -970,11 +994,27 @@ remove_dbtablespaces(Oid db_id)
 		}
 
 		if (!rmtree(dstpath, true))
-		{
 			ereport(WARNING,
 					(errmsg("could not remove database directory \"%s\"",
-							dstpath),
-					 errhint("Look in the postmaster's stderr log for more information.")));
+							dstpath)));
+
+		/* Record the filesystem change in XLOG */
+		{
+			xl_dbase_drop_rec xlrec;
+			XLogRecData rdata[2];
+
+			xlrec.db_id = db_id;
+			rdata[0].buffer = InvalidBuffer;
+			rdata[0].data = (char *) &xlrec;
+			rdata[0].len = offsetof(xl_dbase_drop_rec, dir_path);
+			rdata[0].next = &(rdata[1]);
+
+			rdata[1].buffer = InvalidBuffer;
+			rdata[1].data = (char *) dstpath;
+			rdata[1].len = strlen(dstpath) + 1;
+			rdata[1].next = NULL;
+
+			(void) XLogInsert(RM_DBASE_ID, XLOG_DBASE_DROP, rdata);
 		}
 
 		pfree(dstpath);
@@ -1063,3 +1103,105 @@ get_database_name(Oid dbid)
 
 	return result;
 }
+
+/*
+ * DATABASE resource manager's routines
+ */
+void
+dbase_redo(XLogRecPtr lsn, XLogRecord *record)
+{
+	uint8		info = record->xl_info & ~XLR_INFO_MASK;
+
+	if (info == XLOG_DBASE_CREATE)
+	{
+		xl_dbase_create_rec *xlrec = (xl_dbase_create_rec *) XLogRecGetData(record);
+		char	   *dst_path = xlrec->src_path + strlen(xlrec->src_path) + 1;
+		struct stat st;
+#ifndef WIN32
+		char		buf[2 * MAXPGPATH + 100];
+#endif
+
+		/*
+		 * Our theory for replaying a CREATE is to forcibly drop the target
+		 * subdirectory if present, then re-copy the source data.  This
+		 * may be more work than needed, but it is simple to implement.
+		 */
+		if (stat(dst_path, &st) == 0 && S_ISDIR(st.st_mode))
+		{
+			if (!rmtree(dst_path, true))
+				ereport(WARNING,
+						(errmsg("could not remove database directory \"%s\"",
+								dst_path)));
+		}
+
+#ifndef WIN32
+		/*
+		 * Copy this subdirectory to the new location
+		 *
+		 * XXX use of cp really makes this code pretty grotty, particularly
+		 * with respect to lack of ability to report errors well.  Someday
+		 * rewrite to do it for ourselves.
+		 */
+
+		/* We might need to use cp -R one day for portability */
+		snprintf(buf, sizeof(buf), "cp -r '%s' '%s'",
+				 xlrec->src_path, dst_path);
+		if (system(buf) != 0)
+			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							/* WIN32 */
+		if (copydir(xlrec->src_path, dst_path) != 0)
+		{
+			/* copydir should already have given details of its troubles */
+			ereport(ERROR,
+					(errmsg("could not initialize database directory")));
+		}
+#endif   /* WIN32 */
+	}
+	else if (info == XLOG_DBASE_DROP)
+	{
+		xl_dbase_drop_rec *xlrec = (xl_dbase_drop_rec *) XLogRecGetData(record);
+
+		/* Drop pages for this database that are in the shared buffer cache */
+		DropBuffers(xlrec->db_id);
+
+		if (!rmtree(xlrec->dir_path, true))
+			ereport(WARNING,
+					(errmsg("could not remove database directory \"%s\"",
+							xlrec->dir_path)));
+	}
+	else
+		elog(PANIC, "dbase_redo: unknown op code %u", info);
+}
+
+void
+dbase_undo(XLogRecPtr lsn, XLogRecord *record)
+{
+	elog(PANIC, "dbase_undo: unimplemented");
+}
+
+void
+dbase_desc(char *buf, uint8 xl_info, char *rec)
+{
+	uint8		info = xl_info & ~XLR_INFO_MASK;
+
+	if (info == XLOG_DBASE_CREATE)
+	{
+		xl_dbase_create_rec *xlrec = (xl_dbase_create_rec *) rec;
+		char	   *dst_path = xlrec->src_path + strlen(xlrec->src_path) + 1;
+
+		sprintf(buf + strlen(buf), "create db: %u copy \"%s\" to \"%s\"",
+				xlrec->db_id, xlrec->src_path, dst_path);
+	}
+	else if (info == XLOG_DBASE_DROP)
+	{
+		xl_dbase_drop_rec *xlrec = (xl_dbase_drop_rec *) rec;
+
+		sprintf(buf + strlen(buf), "drop db: %u directory: \"%s\"",
+				xlrec->db_id, xlrec->dir_path);
+	}
+	else
+		strcat(buf, "UNKNOWN");
+}
diff --git a/src/backend/commands/tablespace.c b/src/backend/commands/tablespace.c
index 15fe8392882..3b44ebb19f4 100644
--- a/src/backend/commands/tablespace.c
+++ b/src/backend/commands/tablespace.c
@@ -45,7 +45,7 @@
  *
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/commands/tablespace.c,v 1.9 2004/08/29 05:06:41 momjian Exp $
+ *	  $PostgreSQL: pgsql/src/backend/commands/tablespace.c,v 1.10 2004/08/29 21:08:47 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -73,6 +73,7 @@
 #include "utils/syscache.h"
 
 
+static bool remove_tablespace_directories(Oid tablespaceoid, bool redo);
 static void set_short_version(const char *path);
 static bool directory_is_empty(const char *path);
 
@@ -89,6 +90,13 @@ static bool directory_is_empty(const char *path);
  * isRedo indicates that we are creating an object during WAL replay;
  * we can skip doing locking in that case (and should do so to avoid
  * any possible problems with pg_tablespace not being valid).
+ *
+ * Also, when isRedo is true, we will cope with the possibility of the
+ * tablespace not being there either --- this could happen if we are
+ * replaying an operation on a table in a subsequently-dropped tablespace.
+ * We handle this by making a directory in the place where the tablespace
+ * symlink would normally be.  This isn't an exact replay of course, but
+ * it's the best we can do given the available information.
  */
 void
 TablespaceCreateDbspace(Oid spcNode, Oid dbNode, bool isRedo)
@@ -137,10 +145,29 @@ TablespaceCreateDbspace(Oid spcNode, Oid dbNode, bool isRedo)
 			{
 				/* OK, go for it */
 				if (mkdir(dir, S_IRWXU) < 0)
-					ereport(ERROR,
-							(errcode_for_file_access(),
-						  errmsg("could not create directory \"%s\": %m",
-								 dir)));
+				{
+					char   *parentdir;
+
+					if (errno != ENOENT || !isRedo)
+						ereport(ERROR,
+								(errcode_for_file_access(),
+								 errmsg("could not create directory \"%s\": %m",
+										dir)));
+					/* Try to make parent directory too */
+					parentdir = pstrdup(dir);
+					get_parent_directory(parentdir);
+					if (mkdir(parentdir, S_IRWXU) < 0)
+						ereport(ERROR,
+								(errcode_for_file_access(),
+								 errmsg("could not create directory \"%s\": %m",
+										parentdir)));
+					pfree(parentdir);
+					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 */
@@ -282,11 +309,7 @@ CreateTableSpace(CreateTableSpaceStmt *stmt)
 
 	tuple = heap_formtuple(rel->rd_att, values, nulls);
 
-	tablespaceoid = newoid();
-
-	HeapTupleSetOid(tuple, tablespaceoid);
-
-	simple_heap_insert(rel, tuple);
+	tablespaceoid = simple_heap_insert(rel, tuple);
 
 	CatalogUpdateIndexes(rel, tuple);
 
@@ -332,10 +355,30 @@ CreateTableSpace(CreateTableSpaceStmt *stmt)
 				 errmsg("could not create symbolic link \"%s\": %m",
 						linkloc)));
 
+	/* Record the filesystem change in XLOG */
+	{
+		xl_tblspc_create_rec xlrec;
+		XLogRecData rdata[2];
+
+		xlrec.ts_id = tablespaceoid;
+		rdata[0].buffer = InvalidBuffer;
+		rdata[0].data = (char *) &xlrec;
+		rdata[0].len = offsetof(xl_tblspc_create_rec, ts_path);
+		rdata[0].next = &(rdata[1]);
+
+		rdata[1].buffer = InvalidBuffer;
+		rdata[1].data = (char *) location;
+		rdata[1].len = strlen(location) + 1;
+		rdata[1].next = NULL;
+
+		(void) XLogInsert(RM_TBLSPC_ID, XLOG_TBLSPC_CREATE, rdata);
+	}
+
 	pfree(linkloc);
 	pfree(location);
 
-	heap_close(rel, RowExclusiveLock);
+	/* We keep the lock on pg_tablespace until commit */
+	heap_close(rel, NoLock);
 
 #else							/* !HAVE_SYMLINK */
 	ereport(ERROR,
@@ -358,11 +401,7 @@ DropTableSpace(DropTableSpaceStmt *stmt)
 	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");
@@ -404,7 +443,63 @@ DropTableSpace(DropTableSpaceStmt *stmt)
 		aclcheck_error(ACLCHECK_NO_PRIV, ACL_KIND_TABLESPACE,
 					   tablespacename);
 
-	location = (char *) palloc(strlen(DataDir) + 16 + 10 + 1);
+	/*
+	 * Remove the pg_tablespace tuple (this will roll back if we fail below)
+	 */
+	simple_heap_delete(rel, &tuple->t_self);
+
+	heap_endscan(scandesc);
+
+	/*
+	 * Try to remove the physical infrastructure
+	 */
+	if (!remove_tablespace_directories(tablespaceoid, false))
+		ereport(ERROR,
+				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+				 errmsg("tablespace \"%s\" is not empty",
+						tablespacename)));
+
+	/* Record the filesystem change in XLOG */
+	{
+		xl_tblspc_drop_rec xlrec;
+		XLogRecData rdata[1];
+
+		xlrec.ts_id = tablespaceoid;
+		rdata[0].buffer = InvalidBuffer;
+		rdata[0].data = (char *) &xlrec;
+		rdata[0].len = sizeof(xl_tblspc_drop_rec);
+		rdata[0].next = NULL;
+
+		(void) XLogInsert(RM_TBLSPC_ID, XLOG_TBLSPC_DROP, rdata);
+	}
+
+	/* We keep the lock on pg_tablespace until commit */
+	heap_close(rel, NoLock);
+
+#else							/* !HAVE_SYMLINK */
+	ereport(ERROR,
+			(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+			 errmsg("tablespaces are not supported on this platform")));
+#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(strlen(DataDir) + 11 + 10 + 1);
 	sprintf(location, "%s/pg_tblspc/%u", DataDir, tablespaceoid);
 
 	/*
@@ -422,10 +517,17 @@ DropTableSpace(DropTableSpaceStmt *stmt)
 	 */
 	dirdesc = AllocateDir(location);
 	if (dirdesc == NULL)
+	{
+		if (redo && errno == ENOENT)
+		{
+			pfree(location);
+			return true;
+		}
 		ereport(ERROR,
 				(errcode_for_file_access(),
 				 errmsg("could not open directory \"%s\": %m",
 						location)));
+	}
 
 	errno = 0;
 	while ((de = readdir(dirdesc)) != NULL)
@@ -444,10 +546,10 @@ DropTableSpace(DropTableSpaceStmt *stmt)
 
 		/* 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)));
+		{
+			FreeDir(dirdesc);
+			return false;
+		}
 
 		/* Do the real deed */
 		if (rmdir(subfile) < 0)
@@ -457,6 +559,7 @@ DropTableSpace(DropTableSpaceStmt *stmt)
 							subfile)));
 
 		pfree(subfile);
+		errno = 0;
 	}
 #ifdef WIN32
 
@@ -475,54 +578,51 @@ DropTableSpace(DropTableSpaceStmt *stmt)
 	FreeDir(dirdesc);
 
 	/*
-	 * Okay, try to unlink PG_VERSION and then remove the symlink.
+	 * 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)
-		ereport(ERROR,
-				(errcode_for_file_access(),
-				 errmsg("could not unlink file \"%s\": %m",
-						subfile)));
-
-#ifndef WIN32
-	if (unlink(location) < 0)
-		ereport(ERROR,
-				(errcode_for_file_access(),
-				 errmsg("could not unlink symbolic link \"%s\": %m",
-						location)));
-#else
-	/* The junction is a directory, not a file */
-	if (rmdir(location) < 0)
-		ereport(ERROR,
-				(errcode_for_file_access(),
-				 errmsg("could not remove junction dir \"%s\": %m",
-						location)));
-#endif
+	{
+		if (errno != ENOENT)
+			ereport(ERROR,
+					(errcode_for_file_access(),
+					 errmsg("could not unlink file \"%s\": %m",
+							subfile)));
+	}
 
 	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.
+	 * 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.
 	 */
-	simple_heap_delete(rel, &tuple->t_self);
-
-	heap_endscan(scandesc);
+	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 unlink symbolic link \"%s\": %m",
+							location)));
+	}
 
-	heap_close(rel, ExclusiveLock);
+	pfree(location);
 
-#else							/* !HAVE_SYMLINK */
-	ereport(ERROR,
-			(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
-			 errmsg("tablespaces are not supported on this platform")));
-#endif   /* HAVE_SYMLINK */
+	return true;
 }
 
-
 /*
  * write out the PG_VERSION file in the specified directory
  */
@@ -843,3 +943,88 @@ AlterTableSpaceOwner(const char *name, AclId newOwnerSysId)
 	heap_endscan(scandesc);
 	heap_close(rel, NoLock);
 }
+
+/*
+ * TABLESPACE resource manager's routines
+ */
+void
+tblspc_redo(XLogRecPtr lsn, XLogRecord *record)
+{
+	uint8		info = record->xl_info & ~XLR_INFO_MASK;
+
+	if (info == XLOG_TBLSPC_CREATE)
+	{
+		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 */
+		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)
+		{
+			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)
+	{
+		xl_tblspc_drop_rec *xlrec = (xl_tblspc_drop_rec *) XLogRecGetData(record);
+
+		if (!remove_tablespace_directories(xlrec->ts_id, true))
+			ereport(ERROR,
+					(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+					 errmsg("tablespace %u is not empty",
+							xlrec->ts_id)));
+	}
+	else
+		elog(PANIC, "tblspc_redo: unknown op code %u", info);
+}
+
+void
+tblspc_undo(XLogRecPtr lsn, XLogRecord *record)
+{
+	elog(PANIC, "tblspc_undo: unimplemented");
+}
+
+void
+tblspc_desc(char *buf, uint8 xl_info, char *rec)
+{
+	uint8		info = xl_info & ~XLR_INFO_MASK;
+
+	if (info == XLOG_TBLSPC_CREATE)
+	{
+		xl_tblspc_create_rec *xlrec = (xl_tblspc_create_rec *) rec;
+
+		sprintf(buf + strlen(buf), "create ts: %u \"%s\"",
+				xlrec->ts_id, xlrec->ts_path);
+	}
+	else if (info == XLOG_TBLSPC_DROP)
+	{
+		xl_tblspc_drop_rec *xlrec = (xl_tblspc_drop_rec *) rec;
+
+		sprintf(buf + strlen(buf), "drop ts: %u",
+				xlrec->ts_id);
+	}
+	else
+		strcat(buf, "UNKNOWN");
+}
diff --git a/src/include/access/rmgr.h b/src/include/access/rmgr.h
index d43f6fdcaf1..b01f8058873 100644
--- a/src/include/access/rmgr.h
+++ b/src/include/access/rmgr.h
@@ -3,7 +3,7 @@
  *
  * Resource managers definition
  *
- * $PostgreSQL: pgsql/src/include/access/rmgr.h,v 1.12 2004/08/23 23:22:45 tgl Exp $
+ * $PostgreSQL: pgsql/src/include/access/rmgr.h,v 1.13 2004/08/29 21:08:47 tgl Exp $
  */
 #ifndef RMGR_H
 #define RMGR_H
@@ -17,6 +17,8 @@ typedef uint8 RmgrId;
 #define RM_XACT_ID				1
 #define RM_SMGR_ID				2
 #define RM_CLOG_ID				3
+#define RM_DBASE_ID				4
+#define RM_TBLSPC_ID			5
 #define RM_HEAP_ID				10
 #define RM_BTREE_ID				11
 #define RM_HASH_ID				12
diff --git a/src/include/commands/dbcommands.h b/src/include/commands/dbcommands.h
index f036a38247d..d684511a297 100644
--- a/src/include/commands/dbcommands.h
+++ b/src/include/commands/dbcommands.h
@@ -1,21 +1,41 @@
 /*-------------------------------------------------------------------------
  *
  * dbcommands.h
- *
+ *		Database management commands (create/drop database).
  *
  *
  * Portions Copyright (c) 1996-2004, PostgreSQL Global Development Group
  * Portions Copyright (c) 1994, Regents of the University of California
  *
- * $PostgreSQL: pgsql/src/include/commands/dbcommands.h,v 1.33 2004/08/29 04:13:05 momjian Exp $
+ * $PostgreSQL: pgsql/src/include/commands/dbcommands.h,v 1.34 2004/08/29 21:08:47 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
 #ifndef DBCOMMANDS_H
 #define DBCOMMANDS_H
 
+#include "access/xlog.h"
 #include "nodes/parsenodes.h"
 
+/* XLOG stuff */
+#define XLOG_DBASE_CREATE		0x00
+#define XLOG_DBASE_DROP			0x10
+
+typedef struct xl_dbase_create_rec
+{
+	/* Records copying of a single subdirectory incl. contents */
+	Oid			db_id;
+	char		src_path[1];	/* VARIABLE LENGTH STRING */
+	/* dst_path follows src_path */
+} xl_dbase_create_rec;
+
+typedef struct xl_dbase_drop_rec
+{
+	/* Records dropping of a single subdirectory incl. contents */
+	Oid			db_id;
+	char		dir_path[1];	/* VARIABLE LENGTH STRING */
+} xl_dbase_drop_rec;
+
 extern void createdb(const CreatedbStmt *stmt);
 extern void dropdb(const char *dbname);
 extern void RenameDatabase(const char *oldname, const char *newname);
@@ -25,4 +45,8 @@ extern void AlterDatabaseOwner(const char *dbname, AclId newOwnerSysId);
 extern Oid	get_database_oid(const char *dbname);
 extern char *get_database_name(Oid dbid);
 
+extern void dbase_redo(XLogRecPtr lsn, XLogRecord *rptr);
+extern void dbase_undo(XLogRecPtr lsn, XLogRecord *rptr);
+extern void dbase_desc(char *buf, uint8 xl_info, char *rec);
+
 #endif   /* DBCOMMANDS_H */
diff --git a/src/include/commands/tablespace.h b/src/include/commands/tablespace.h
index 42093c51505..8aba919ab3a 100644
--- a/src/include/commands/tablespace.h
+++ b/src/include/commands/tablespace.h
@@ -1,32 +1,49 @@
 /*-------------------------------------------------------------------------
  *
  * tablespace.h
- *	  prototypes for tablespace.c.
+ *		Tablespace management commands (create/drop tablespace).
  *
  *
  * 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.3 2004/07/11 19:52:52 tgl Exp $
+ * $PostgreSQL: pgsql/src/include/commands/tablespace.h,v 1.4 2004/08/29 21:08:47 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
 #ifndef TABLESPACE_H
 #define TABLESPACE_H
 
+#include "access/xlog.h"
 #include "nodes/parsenodes.h"
 
-extern void CreateTableSpace(CreateTableSpaceStmt *stmt);
+/* XLOG stuff */
+#define XLOG_TBLSPC_CREATE		0x00
+#define XLOG_TBLSPC_DROP		0x10
+
+typedef struct xl_tblspc_create_rec
+{
+	Oid			ts_id;
+	char		ts_path[1];		/* VARIABLE LENGTH STRING */
+} xl_tblspc_create_rec;
+
+typedef struct xl_tblspc_drop_rec
+{
+	Oid			ts_id;
+} xl_tblspc_drop_rec;
 
+extern void CreateTableSpace(CreateTableSpaceStmt *stmt);
 extern void DropTableSpace(DropTableSpaceStmt *stmt);
+extern void RenameTableSpace(const char *oldname, const char *newname);
+extern void AlterTableSpaceOwner(const char *name, AclId newOwnerSysId);
 
 extern void TablespaceCreateDbspace(Oid spcNode, Oid dbNode, bool isRedo);
 
 extern Oid	get_tablespace_oid(const char *tablespacename);
-
 extern char *get_tablespace_name(Oid spc_oid);
 
-extern void RenameTableSpace(const char *oldname, const char *newname);
-extern void AlterTableSpaceOwner(const char *name, AclId newOwnerSysId);
+extern void tblspc_redo(XLogRecPtr lsn, XLogRecord *rptr);
+extern void tblspc_undo(XLogRecPtr lsn, XLogRecord *rptr);
+extern void tblspc_desc(char *buf, uint8 xl_info, char *rec);
 
 #endif   /* TABLESPACE_H */
diff --git a/src/include/port.h b/src/include/port.h
index cd9225748e8..e3e4eda4569 100644
--- a/src/include/port.h
+++ b/src/include/port.h
@@ -6,7 +6,7 @@
  * Portions Copyright (c) 1996-2004, PostgreSQL Global Development Group
  * Portions Copyright (c) 1994, Regents of the University of California
  *
- * $PostgreSQL: pgsql/src/include/port.h,v 1.56 2004/08/29 05:06:55 momjian Exp $
+ * $PostgreSQL: pgsql/src/include/port.h,v 1.57 2004/08/29 21:08:48 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -51,6 +51,7 @@ extern void get_pkglib_path(const char *my_exec_path, char *ret_path);
 extern void get_locale_path(const char *my_exec_path, char *ret_path);
 extern void set_pglocale_pgservice(const char *argv0, const char *app);
 extern bool get_home_path(char *ret_path);
+extern void get_parent_directory(char *path);
 
 /*
  *	is_absolute_path
diff --git a/src/port/path.c b/src/port/path.c
index dae4eeab099..8017417b708 100644
--- a/src/port/path.c
+++ b/src/port/path.c
@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/port/path.c,v 1.33 2004/08/29 05:07:02 momjian Exp $
+ *	  $PostgreSQL: pgsql/src/port/path.c,v 1.34 2004/08/29 21:08:48 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -324,6 +324,39 @@ get_locale_path(const char *my_exec_path, char *ret_path)
 }
 
 
+/*
+ *	get_home_path
+ */
+bool
+get_home_path(char *ret_path)
+{
+	if (getenv(HOMEDIR) == NULL)
+	{
+		*ret_path = '\0';
+		return false;
+	}
+	else
+	{
+		StrNCpy(ret_path, getenv(HOMEDIR), MAXPGPATH);
+		canonicalize_path(ret_path);
+		return true;
+	}
+}
+
+
+/*
+ * get_parent_directory
+ *
+ * Modify the given string in-place to name the parent directory of the
+ * named file.
+ */
+void
+get_parent_directory(char *path)
+{
+	trim_directory(path);
+	trim_trailing_separator(path);
+}
+
 
 /*
  *	set_pglocale_pgservice
@@ -374,27 +407,6 @@ set_pglocale_pgservice(const char *argv0, const char *app)
 
 
 /*
- *	get_include_path
- */
-bool
-get_home_path(char *ret_path)
-{
-	if (getenv(HOMEDIR) == NULL)
-	{
-		*ret_path = '\0';
-		return false;
-	}
-	else
-	{
-		StrNCpy(ret_path, getenv(HOMEDIR), MAXPGPATH);
-		canonicalize_path(ret_path);
-		return true;
-	}
-}
-
-
-
-/*
  *	make_relative - adjust path to be relative to bin/
  */
 static void