Make DROP DATABASE command generate less WAL records.

Previously DROP DATABASE generated as many XLOG_DBASE_DROP WAL records
as the number of tablespaces that the database to drop uses. This caused
the scans of shared_buffers as many times as the number of the tablespaces
during recovery because WAL replay of one XLOG_DBASE_DROP record needs
that full scan. This could make the recovery time longer especially
when shared_buffers is large.

This commit changes DROP DATABASE so that it generates only one
XLOG_DBASE_DROP record, and registers the information of all the tablespaces
into it. Then, WAL replay of XLOG_DBASE_DROP record needs full scan of
shared_buffers only once, and which may improve the recovery performance.

Author: Fujii Masao
Reviewed-by: Kirk Jamison, Simon Riggs
Discussion: https://postgr.es/m/CAHGQGwF8YwNH0ZaL+2wjZPkj+ji9UhC+Z4ScnG97WKtVY5L9iw@mail.gmail.com

src/backend/access/rmgrdesc/dbasedesc.c

@@ -35,9 +35,12 @@ dbase_desc(StringInfo buf, XLogReaderState *record)
 	else if (info == XLOG_DBASE_DROP)
 	{
 		xl_dbase_drop_rec *xlrec = (xl_dbase_drop_rec *) rec;
+		int		i;
 
-		appendStringInfo(buf, "dir %u/%u",
+		appendStringInfo(buf, "dir");
-						 xlrec->tablespace_id, xlrec->db_id);
+		for (i = 0; i < xlrec->ntablespaces; i++)
+			appendStringInfo(buf, " %u/%u",
+							 xlrec->tablespace_ids[i], xlrec->db_id);
 	}
 }
 

src/backend/commands/dbcommands.c

@@ -1411,10 +1411,11 @@ movedb(const char *dbname, const char *tblspcname)
 		xl_dbase_drop_rec xlrec;
 
 		xlrec.db_id = db_id;
-		xlrec.tablespace_id = src_tblspcoid;
+		xlrec.ntablespaces = 1;
 
 		XLogBeginInsert();
 		XLogRegisterData((char *) &xlrec, sizeof(xl_dbase_drop_rec));
+		XLogRegisterData((char *) &src_tblspcoid, sizeof(Oid));
 
 		(void) XLogInsert(RM_DBASE_ID,
 						  XLOG_DBASE_DROP | XLR_SPECIAL_REL_UPDATE);
@@ -1946,6 +1947,11 @@ remove_dbtablespaces(Oid db_id)
 	Relation	rel;
 	TableScanDesc scan;
 	HeapTuple	tuple;
+	List		*ltblspc = NIL;
+	ListCell	*cell;
+	int		ntblspc;
+	int		i;
+	Oid		*tablespace_ids;
 
 	rel = table_open(TableSpaceRelationId, AccessShareLock);
 	scan = table_beginscan_catalog(rel, 0, NULL);
@@ -1974,23 +1980,41 @@ remove_dbtablespaces(Oid db_id)
 					(errmsg("some useless files may be left behind in old database directory \"%s\"",
 							dstpath)));
 
-		/* Record the filesystem change in XLOG */
+		ltblspc = lappend_oid(ltblspc, dsttablespace);
-		{
+		pfree(dstpath);
-			xl_dbase_drop_rec xlrec;
+	}
 
-			xlrec.db_id = db_id;
+	ntblspc = list_length(ltblspc);
-			xlrec.tablespace_id = dsttablespace;
+	if (ntblspc == 0)
+	{
+		table_endscan(scan);
+		table_close(rel, AccessShareLock);
+		return;
+	}
 
-			XLogBeginInsert();
+	tablespace_ids = (Oid *) palloc(ntblspc * sizeof(Oid));
-			XLogRegisterData((char *) &xlrec, sizeof(xl_dbase_drop_rec));
+	i = 0;
+	foreach(cell, ltblspc)
+		tablespace_ids[i++] = lfirst_oid(cell);
 
-			(void) XLogInsert(RM_DBASE_ID,
+	/* Record the filesystem change in XLOG */
-							  XLOG_DBASE_DROP | XLR_SPECIAL_REL_UPDATE);
+	{
-		}
+		xl_dbase_drop_rec xlrec;
 
-		pfree(dstpath);
+		xlrec.db_id = db_id;
+		xlrec.ntablespaces = ntblspc;
+
+		XLogBeginInsert();
+		XLogRegisterData((char *) &xlrec, MinSizeOfDbaseDropRec);
+		XLogRegisterData((char *) tablespace_ids, ntblspc * sizeof(Oid));
+
+		(void) XLogInsert(RM_DBASE_ID,
+						  XLOG_DBASE_DROP | XLR_SPECIAL_REL_UPDATE);
 	}
 
+	list_free(ltblspc);
+	pfree(tablespace_ids);
+
 	table_endscan(scan);
 	table_close(rel, AccessShareLock);
 }
@@ -2197,8 +2221,7 @@ dbase_redo(XLogReaderState *record)
 	{
 		xl_dbase_drop_rec *xlrec = (xl_dbase_drop_rec *) XLogRecGetData(record);
 		char	   *dst_path;
-
+		int			i;
-		dst_path = GetDatabasePath(xlrec->db_id, xlrec->tablespace_id);
 
 		if (InHotStandby)
 		{
@@ -2228,11 +2251,17 @@ dbase_redo(XLogReaderState *record)
 		/* Clean out the xlog relcache too */
 		XLogDropDatabase(xlrec->db_id);
 
-		/* And remove the physical files */
+		for (i = 0; i < xlrec->ntablespaces; i++)
-		if (!rmtree(dst_path, true))
+		{
-			ereport(WARNING,
+			dst_path = GetDatabasePath(xlrec->db_id, xlrec->tablespace_ids[i]);
-					(errmsg("some useless files may be left behind in old database directory \"%s\"",
+
-							dst_path)));
+			/* And remove the physical files */
+			if (!rmtree(dst_path, true))
+				ereport(WARNING,
+						(errmsg("some useless files may be left behind in old database directory \"%s\"",
+								dst_path)));
+			pfree(dst_path);
+		}
 
 		if (InHotStandby)
 		{

src/include/commands/dbcommands_xlog.h

@@ -32,10 +32,11 @@ typedef struct xl_dbase_create_rec
 
 typedef struct xl_dbase_drop_rec
 {
-	/* Records dropping of a single subdirectory incl. contents */
 	Oid			db_id;
-	Oid			tablespace_id;
+	int			ntablespaces;		/* number of tablespace IDs */
+	Oid			tablespace_ids[FLEXIBLE_ARRAY_MEMBER];
 } xl_dbase_drop_rec;
+#define MinSizeOfDbaseDropRec offsetof(xl_dbase_drop_rec, tablespace_ids)
 
 extern void dbase_redo(XLogReaderState *rptr);
 extern void dbase_desc(StringInfo buf, XLogReaderState *rptr);