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

diff --git a/src/backend/access/rmgrdesc/dbasedesc.c b/src/backend/access/rmgrdesc/dbasedesc.c
index c7d60ce10d6845e71cbce6ba5275df5ae55659cc..d08c57587296b5ccc98eb76dae750932c3338d6d 100644 (file)
--- a/src/backend/access/rmgrdesc/dbasedesc.c
+++ b/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",
-                        xlrec->tablespace_id, xlrec->db_id);
+       appendStringInfo(buf, "dir");
+       for (i = 0; i < xlrec->ntablespaces; i++)
+           appendStringInfo(buf, " %u/%u",
+                            xlrec->tablespace_ids[i], xlrec->db_id);
    }
 }
 

diff --git a/src/backend/commands/dbcommands.c b/src/backend/commands/dbcommands.c
index 446813f0f0b2a0530c0bdc9b0d36585da9019ab1..a70e75a219e3a0c7d18a18845129a6a89886841b 100644 (file)
--- a/src/backend/commands/dbcommands.c
+++ b/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 */
-       {
-           xl_dbase_drop_rec xlrec;
+       ltblspc = lappend_oid(ltblspc, dsttablespace);
+       pfree(dstpath);
+   }
 
-           xlrec.db_id = db_id;
-           xlrec.tablespace_id = dsttablespace;
+   ntblspc = list_length(ltblspc);
+   if (ntblspc == 0)
+   {
+       table_endscan(scan);
+       table_close(rel, AccessShareLock);
+       return;
+   }
 
-           XLogBeginInsert();
-           XLogRegisterData((char *) &xlrec, sizeof(xl_dbase_drop_rec));
+   tablespace_ids = (Oid *) palloc(ntblspc * sizeof(Oid));
+   i = 0;
+   foreach(cell, ltblspc)
+       tablespace_ids[i++] = lfirst_oid(cell);
 
-           (void) XLogInsert(RM_DBASE_ID,
-                             XLOG_DBASE_DROP | XLR_SPECIAL_REL_UPDATE);
-       }
+   /* Record the filesystem change in XLOG */
+   {
+       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;
-
-       dst_path = GetDatabasePath(xlrec->db_id, xlrec->tablespace_id);
+       int         i;
 
        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 */
-       if (!rmtree(dst_path, true))
-           ereport(WARNING,
-                   (errmsg("some useless files may be left behind in old database directory \"%s\"",
-                           dst_path)));
+       for (i = 0; i < xlrec->ntablespaces; i++)
+       {
+           dst_path = GetDatabasePath(xlrec->db_id, xlrec->tablespace_ids[i]);
+
+           /* 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)
        {

diff --git a/src/include/commands/dbcommands_xlog.h b/src/include/commands/dbcommands_xlog.h
index 46be8a615accc6a9fb05e157ee637b248188f7b0..4e8a442c89f6330527d73e8f95f9c4057ad6af27 100644 (file)
--- a/src/include/commands/dbcommands_xlog.h
+++ b/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);