database/postgres
database/postgres
1
0
Fork 0
mirror of https://github.com/postgres/postgres.git synced 2025-05-02 11:44:50 +03:00
Code

Make DROP DATABASE command generate less WAL records.

Browse Source 
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
This commit is contained in:
Fujii Masao 2019-11-21 21:10:37 +09:00
parent 30840c92ac
commit e6d8069522
3 changed files with 59 additions and 26 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
src/backend/access/rmgrdesc/dbasedesc.c
View File

@ -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);
}
}

43
src/backend/commands/dbcommands.c
View File

@ -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,22 +1980,40 @@ remove_dbtablespaces(Oid db_id)
(errmsg("some useless files may be left behind in old database directory \"%s\"",
dstpath)));
ltblspc = lappend_oid(ltblspc, dsttablespace);
pfree(dstpath);
}
ntblspc = list_length(ltblspc);
if (ntblspc == 0)
{
table_endscan(scan);
table_close(rel, AccessShareLock);
return;
}
tablespace_ids = (Oid *) palloc(ntblspc * sizeof(Oid));
i = 0;
foreach(cell, ltblspc)
tablespace_ids[i++] = lfirst_oid(cell);
/* Record the filesystem change in XLOG */
{
xl_dbase_drop_rec xlrec;
xlrec.db_id = db_id;
xlrec.tablespace_id = dsttablespace;
xlrec.ntablespaces = ntblspc;
XLogBeginInsert();
XLogRegisterData((char *) &xlrec, sizeof(xl_dbase_drop_rec));
XLogRegisterData((char *) &xlrec, MinSizeOfDbaseDropRec);
XLogRegisterData((char *) tablespace_ids, ntblspc * sizeof(Oid));
(void) XLogInsert(RM_DBASE_ID,
XLOG_DBASE_DROP | XLR_SPECIAL_REL_UPDATE);
}
pfree(dstpath);
}
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);
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)
{

5
src/include/commands/dbcommands_xlog.h
View File

@ -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);