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.

doc/src/sgml/backup.sgml

 <!--
-$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,17 +902,9 @@ 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
@@ -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>

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},

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");
+}

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

src/include/commands/dbcommands.h

 /*-------------------------------------------------------------------------
  *
  * 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 */

src/include/commands/tablespace.h

 /*-------------------------------------------------------------------------
  *
  * 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 */

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

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
@@ -373,27 +406,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/
  */