WAL must log CREATE and DROP DATABASE operations *without* using any

explicit paths, so that the log can be replayed in a data directory
with a different absolute path than the original had.  To avoid forcing
initdb in the 8.0 branch, continue to accept the old WAL log record
types; they will never again be generated however, and the code can be
dropped after the next forced initdb.  Per report from Oleg Bartunov.
We still need to think about what it really means to WAL-log CREATE
TABLESPACE commands: we more or less have to put the absolute path
into those, but how to replay in a different context??

src/backend/commands/dbcommands.c

@@ -15,7 +15,7 @@
  *
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/commands/dbcommands.c,v 1.154 2005/03/12 21:33:55 tgl Exp $
+ *	  $PostgreSQL: pgsql/src/backend/commands/dbcommands.c,v 1.155 2005/03/23 00:03:28 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -438,23 +438,17 @@ createdb(const CreatedbStmt *stmt)
 		/* Record the filesystem change in XLOG */
 		{
 			xl_dbase_create_rec xlrec;
-			XLogRecData rdata[3];
+			XLogRecData rdata[1];
 
 			xlrec.db_id = dboid;
+			xlrec.tablespace_id = dsttablespace;
+			xlrec.src_db_id = src_dboid;
+			xlrec.src_tablespace_id = srctablespace;
+
 			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;
+			rdata[0].len = sizeof(xl_dbase_create_rec);
+			rdata[0].next = NULL;
 
 			(void) XLogInsert(RM_DBASE_ID, XLOG_DBASE_CREATE, rdata);
 		}
@@ -1076,18 +1070,15 @@ remove_dbtablespaces(Oid db_id)
 		/* Record the filesystem change in XLOG */
 		{
 			xl_dbase_drop_rec xlrec;
-			XLogRecData rdata[2];
+			XLogRecData rdata[1];
 
 			xlrec.db_id = db_id;
+			xlrec.tablespace_id = dsttablespace;
+
 			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;
+			rdata[0].len = sizeof(xl_dbase_drop_rec);
+			rdata[0].next = NULL;
 
 			(void) XLogInsert(RM_DBASE_ID, XLOG_DBASE_DROP, rdata);
 		}
@@ -1190,6 +1181,86 @@ dbase_redo(XLogRecPtr lsn, XLogRecord *record)
 	if (info == XLOG_DBASE_CREATE)
 	{
 		xl_dbase_create_rec *xlrec = (xl_dbase_create_rec *) XLogRecGetData(record);
+		char	   *src_path;
+		char	   *dst_path;
+		struct stat st;
+
+#ifndef WIN32
+		char		buf[2 * MAXPGPATH + 100];
+#endif
+
+		src_path = GetDatabasePath(xlrec->src_db_id, xlrec->src_tablespace_id);
+		dst_path = GetDatabasePath(xlrec->db_id, xlrec->tablespace_id);
+
+		/*
+		 * 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)));
+		}
+
+		/*
+		 * Force dirty buffers out to disk, to ensure source database is
+		 * up-to-date for the copy.  (We really only need to flush buffers for
+		 * the source database, but bufmgr.c provides no API for that.)
+		 */
+		BufferSync();
+
+#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'",
+				 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(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);
+		char	   *dst_path;
+
+		dst_path = GetDatabasePath(xlrec->db_id, xlrec->tablespace_id);
+
+		/*
+		 * Drop pages for this database that are in the shared buffer
+		 * cache
+		 */
+		DropBuffers(xlrec->db_id);
+
+		if (!rmtree(dst_path, true))
+			ereport(WARNING,
+					(errmsg("could not remove database directory \"%s\"",
+							dst_path)));
+	}
+	else if (info == XLOG_DBASE_CREATE_OLD)
+	{
+		xl_dbase_create_rec_old *xlrec = (xl_dbase_create_rec_old *) XLogRecGetData(record);
 		char	   *dst_path = xlrec->src_path + strlen(xlrec->src_path) + 1;
 		struct stat st;
 
@@ -1245,9 +1316,9 @@ dbase_redo(XLogRecPtr lsn, XLogRecord *record)
 		}
 #endif   /* WIN32 */
 	}
-	else if (info == XLOG_DBASE_DROP)
+	else if (info == XLOG_DBASE_DROP_OLD)
 	{
-		xl_dbase_drop_rec *xlrec = (xl_dbase_drop_rec *) XLogRecGetData(record);
+		xl_dbase_drop_rec_old *xlrec = (xl_dbase_drop_rec_old *) XLogRecGetData(record);
 
 		/*
 		 * Drop pages for this database that are in the shared buffer
@@ -1278,14 +1349,29 @@ dbase_desc(char *buf, uint8 xl_info, char *rec)
 	if (info == XLOG_DBASE_CREATE)
 	{
 		xl_dbase_create_rec *xlrec = (xl_dbase_create_rec *) rec;
+
+		sprintf(buf + strlen(buf), "create db: copy dir %u/%u to %u/%u",
+				xlrec->src_db_id, xlrec->src_tablespace_id,
+				xlrec->db_id, xlrec->tablespace_id);
+	}
+	else if (info == XLOG_DBASE_DROP)
+	{
+		xl_dbase_drop_rec *xlrec = (xl_dbase_drop_rec *) rec;
+
+		sprintf(buf + strlen(buf), "drop db: dir %u/%u",
+				xlrec->db_id, xlrec->tablespace_id);
+	}
+	else if (info == XLOG_DBASE_CREATE_OLD)
+	{
+		xl_dbase_create_rec_old *xlrec = (xl_dbase_create_rec_old *) 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)
+	else if (info == XLOG_DBASE_DROP_OLD)
 	{
-		xl_dbase_drop_rec *xlrec = (xl_dbase_drop_rec *) rec;
+		xl_dbase_drop_rec_old *xlrec = (xl_dbase_drop_rec_old *) rec;
 
 		sprintf(buf + strlen(buf), "drop db: %u directory: \"%s\"",
 				xlrec->db_id, xlrec->dir_path);

src/include/commands/dbcommands.h

@@ -7,7 +7,7 @@
  * Portions Copyright (c) 1996-2005, PostgreSQL Global Development Group
  * Portions Copyright (c) 1994, Regents of the University of California
  *
- * $PostgreSQL: pgsql/src/include/commands/dbcommands.h,v 1.36 2004/12/31 22:03:28 pgsql Exp $
+ * $PostgreSQL: pgsql/src/include/commands/dbcommands.h,v 1.37 2005/03/23 00:03:37 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -18,22 +18,47 @@
 #include "nodes/parsenodes.h"
 
 /* XLOG stuff */
-#define XLOG_DBASE_CREATE		0x00
-#define XLOG_DBASE_DROP			0x10
+#define XLOG_DBASE_CREATE_OLD	0x00
+#define XLOG_DBASE_DROP_OLD		0x10
+#define XLOG_DBASE_CREATE		0x20
+#define XLOG_DBASE_DROP			0x30
 
-typedef struct xl_dbase_create_rec
+/*
+ * Note: "old" versions are deprecated and need not be supported beyond 8.0.
+ * Not only are they relatively bulky, but they do the Wrong Thing when a
+ * WAL log is replayed in a data area that's at a different absolute path
+ * than the original.
+ */
+
+typedef struct xl_dbase_create_rec_old
 {
 	/* 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_old;
+
+typedef struct xl_dbase_drop_rec_old
+{
+	/* Records dropping of a single subdirectory incl. contents */
+	Oid			db_id;
+	char		dir_path[1];	/* VARIABLE LENGTH STRING */
+}	xl_dbase_drop_rec_old;
+
+typedef struct xl_dbase_create_rec
+{
+	/* Records copying of a single subdirectory incl. contents */
+	Oid			db_id;
+	Oid			tablespace_id;
+	Oid			src_db_id;
+	Oid			src_tablespace_id;
 }	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 */
+	Oid			tablespace_id;
 }	xl_dbase_drop_rec;
 
 extern void createdb(const CreatedbStmt *stmt);