rdelaage
/
postgresql
mirror of https://git.postgresql.org/git/postgresql.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Implement ALTER DATABASE SET TABLESPACE to move a whole database (or at least 

as much of it as lives in its default tablespace) to a new tablespace.

Guillaume Lelarge, with some help from Bernd Helmle and Tom Lane
This commit is contained in:
Tom Lane 2008-11-07 18:25:07 +00:00
parent 85e2cedf98
commit 6517f377d6
8 changed files with 437 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
doc/src/sgml/ref/alter_database.sgml
View File

@ -1,5 +1,5 @@
<!--
$PostgreSQL: pgsql/doc/src/sgml/ref/alter_database.sgml,v 1.20 2007/09/03 18:46:29 tgl Exp $
$PostgreSQL: pgsql/doc/src/sgml/ref/alter_database.sgml,v 1.21 2008/11/07 18:25:06 tgl Exp $
PostgreSQL documentation
-->
@ -30,6 +30,8 @@ ALTER DATABASE <replaceable class="PARAMETER">name</replaceable> RENAME TO <repl
ALTER DATABASE <replaceable class="PARAMETER">name</replaceable> OWNER TO <replaceable>new_owner</replaceable>
ALTER DATABASE <replaceable class="PARAMETER">name</replaceable> SET TABLESPACE <replaceable class="PARAMETER">new_tablespace</replaceable>
ALTER DATABASE <replaceable class="PARAMETER">name</replaceable> SET <replaceable>configuration_parameter</replaceable> { TO | = } { <replaceable>value</replaceable> | DEFAULT }
ALTER DATABASE <replaceable class="PARAMETER">name</replaceable> SET <replaceable>configuration_parameter</replaceable> FROM CURRENT
ALTER DATABASE <replaceable class="PARAMETER">name</replaceable> RESET <replaceable>configuration_parameter</replaceable>
@ -67,6 +69,15 @@ ALTER DATABASE <replaceable class="PARAMETER">name</replaceable> RESET ALL
(Note that superusers have all these privileges automatically.)
</para>
<para>
The fourth form changes the default tablespace of the database.
Only the database owner or a superuser can do this; you must also have
create privilege for the new tablespace.
This command physically moves any tables or indexes in the database's old
default tablespace to the new tablespace. Note that tables and indexes
in non-default tablespaces are not affected.
</para>
<para>
The remaining forms change the session default for a run-time
configuration variable for a <productname>PostgreSQL</productname>
@ -122,6 +133,15 @@ ALTER DATABASE <replaceable class="PARAMETER">name</replaceable> RESET ALL
</listitem>
</varlistentry>
<varlistentry>
<term><replaceable class="parameter">new_tablespace</replaceable></term>
<listitem>
<para>
The new default tablespace of the database.
</para>
</listitem>
</varlistentry>
<varlistentry>
<term><replaceable>configuration_parameter</replaceable></term>
<term><replaceable>value</replaceable></term>
@ -189,6 +209,7 @@ ALTER DATABASE test SET enable_indexscan TO off;
<member><xref linkend="sql-createdatabase" endterm="sql-createdatabase-title"></member>
<member><xref linkend="sql-dropdatabase" endterm="sql-dropdatabase-title"></member>
<member><xref linkend="sql-set" endterm="sql-set-title"></member>
<member><xref linkend="sql-createtablespace" endterm="sql-createtablespace-title"></member>
</simplelist>
</refsect1>
</refentry>

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

@ -13,7 +13,7 @@
*
*
* IDENTIFICATION
* $PostgreSQL: pgsql/src/backend/commands/dbcommands.c,v 1.215 2008/11/02 01:45:27 tgl Exp $
* $PostgreSQL: pgsql/src/backend/commands/dbcommands.c,v 1.216 2008/11/07 18:25:06 tgl Exp $
*
*-------------------------------------------------------------------------
*/
@ -42,6 +42,7 @@
#include "pgstat.h"
#include "postmaster/bgwriter.h"
#include "storage/bufmgr.h"
#include "storage/fd.h"
#include "storage/lmgr.h"
#include "storage/ipc.h"
#include "storage/procarray.h"
@ -53,6 +54,7 @@
#include "utils/guc.h"
#include "utils/lsyscache.h"
#include "utils/pg_locale.h"
#include "utils/snapmgr.h"
#include "utils/syscache.h"
#include "utils/tqual.h"
@ -63,8 +65,16 @@ typedef struct
Oid dest_dboid; /* DB we are trying to create */
} createdb_failure_params;
typedef struct
{
Oid dest_dboid; /* DB we are trying to move */
Oid dest_tsoid; /* tablespace we are trying to move to */
} movedb_failure_params;
/* non-export function prototypes */
static void createdb_failure_callback(int code, Datum arg);
static void movedb(const char *dbname, const char *tblspcname);
static void movedb_failure_callback(int code, Datum arg);
static bool get_db_info(const char *name, LOCKMODE lockmode,
Oid *dbIdP, Oid *ownerIdP,
int *encodingP, bool *dbIsTemplateP, bool *dbAllowConnP,
@ -934,11 +944,330 @@ RenameDatabase(const char *oldname, const char *newname)
}
/*
* ALTER DATABASE SET TABLESPACE
*/
static void
movedb(const char *dbname, const char *tblspcname)
{
Oid db_id;
Relation pgdbrel;
int notherbackends;
int npreparedxacts;
HeapTuple oldtuple, newtuple;
Oid src_tblspcoid, dst_tblspcoid;
Datum new_record[Natts_pg_database];
bool new_record_nulls[Natts_pg_database];
bool new_record_repl[Natts_pg_database];
ScanKeyData scankey;
SysScanDesc sysscan;
AclResult aclresult;
char *src_dbpath;
char *dst_dbpath;
DIR *dstdir;
struct dirent *xlde;
movedb_failure_params fparms;
/*
* Look up the target database's OID, and get exclusive lock on it. We
* need this to ensure that no new backend starts up in the database while
* we are moving it, and that no one is using it as a CREATE DATABASE
* template or trying to delete it.
*/
pgdbrel = heap_open(DatabaseRelationId, RowExclusiveLock);
if (!get_db_info(dbname, AccessExclusiveLock, &db_id, NULL, NULL,
NULL, NULL, NULL, NULL, &src_tblspcoid, NULL, NULL))
ereport(ERROR,
(errcode(ERRCODE_UNDEFINED_DATABASE),
errmsg("database \"%s\" does not exist", dbname)));
/*
* We actually need a session lock, so that the lock will persist across
* the commit/restart below. (We could almost get away with letting the
* lock be released at commit, except that someone could try to move
* relations of the DB back into the old directory while we rmtree() it.)
*/
LockSharedObjectForSession(DatabaseRelationId, db_id, 0,
AccessExclusiveLock);
/*
* Permission checks
*/
if (!pg_database_ownercheck(db_id, GetUserId()))
aclcheck_error(ACLCHECK_NOT_OWNER, ACL_KIND_DATABASE,
dbname);
/*
* Obviously can't move the tables of my own database
*/
if (db_id == MyDatabaseId)
ereport(ERROR,
(errcode(ERRCODE_OBJECT_IN_USE),
errmsg("cannot change the tablespace of the currently open database")));
/*
* Get tablespace's oid
*/
dst_tblspcoid = get_tablespace_oid(tblspcname);
if (dst_tblspcoid == InvalidOid)
ereport(ERROR,
(errcode(ERRCODE_UNDEFINED_DATABASE),
errmsg("tablespace \"%s\" does not exist", tblspcname)));
/*
* Permission checks
*/
aclresult = pg_tablespace_aclcheck(dst_tblspcoid, GetUserId(),
ACL_CREATE);
if (aclresult != ACLCHECK_OK)
aclcheck_error(aclresult, ACL_KIND_TABLESPACE,
tblspcname);
/*
* pg_global must never be the default tablespace
*/
if (dst_tblspcoid == GLOBALTABLESPACE_OID)
ereport(ERROR,
(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
errmsg("pg_global cannot be used as default tablespace")));
/*
* No-op if same tablespace
*/
if (src_tblspcoid == dst_tblspcoid)
{
heap_close(pgdbrel, NoLock);
UnlockSharedObjectForSession(DatabaseRelationId, db_id, 0,
AccessExclusiveLock);
return;
}
/*
* Check for other backends in the target database. (Because we hold the
* database lock, no new ones can start after this.)
*
* As in CREATE DATABASE, check this after other error conditions.
*/
if (CountOtherDBBackends(db_id, &notherbackends, &npreparedxacts))
ereport(ERROR,
(errcode(ERRCODE_OBJECT_IN_USE),
errmsg("database \"%s\" is being accessed by other users",
dbname),
errdetail_busy_db(notherbackends, npreparedxacts)));
/*
* Get old and new database paths
*/
src_dbpath = GetDatabasePath(db_id, src_tblspcoid);
dst_dbpath = GetDatabasePath(db_id, dst_tblspcoid);
/*
* Force a checkpoint before proceeding. This will force dirty buffers out
* to disk, to ensure source database is up-to-date on disk for the
* copy. FlushDatabaseBuffers() would suffice for that, but we also want
* to process any pending unlink requests. Otherwise, the check for
* existing files in the target directory might fail unnecessarily, not to
* mention that the copy might fail due to source files getting deleted
* under it. On Windows, this also ensures that the bgwriter doesn't hold
* any open files, which would cause rmdir() to fail.
*/
RequestCheckpoint(CHECKPOINT_IMMEDIATE | CHECKPOINT_FORCE | CHECKPOINT_WAIT);
/*
* Check for existence of files in the target directory, i.e., objects of
* this database that are already in the target tablespace. We can't
* allow the move in such a case, because we would need to change those
* relations' pg_class.reltablespace entries to zero, and we don't have
* access to the DB's pg_class to do so.
*/
dstdir = AllocateDir(dst_dbpath);
if (dstdir != NULL)
{
while ((xlde = ReadDir(dstdir, dst_dbpath)) != NULL)
{
if (strcmp(xlde->d_name, ".") == 0 ||
strcmp(xlde->d_name, "..") == 0)
continue;
ereport(ERROR,
(errmsg("some relations of database \"%s\" are already in tablespace \"%s\"",
dbname, tblspcname),
errhint("You must move them back to the database's default tablespace before using this command.")));
}
FreeDir(dstdir);
/*
* The directory exists but is empty.
* We must remove it before using the copydir function.
*/
if (rmdir(dst_dbpath) != 0)
elog(ERROR, "could not remove directory \"%s\": %m",
dst_dbpath);
}
/*
* Use an ENSURE block to make sure we remove the debris if the copy fails
* (eg, due to out-of-disk-space). This is not a 100% solution, because
* of the possibility of failure during transaction commit, but it should
* handle most scenarios.
*/
fparms.dest_dboid = db_id;
fparms.dest_tsoid = dst_tblspcoid;
PG_ENSURE_ERROR_CLEANUP(movedb_failure_callback,
PointerGetDatum(&fparms));
{
/*
* Copy files from the old tablespace to the new one
*/
copydir(src_dbpath, dst_dbpath, false);
/*
* Record the filesystem change in XLOG
*/
{
xl_dbase_create_rec xlrec;
XLogRecData rdata[1];
xlrec.db_id = db_id;
xlrec.tablespace_id = dst_tblspcoid;
xlrec.src_db_id = db_id;
xlrec.src_tablespace_id = src_tblspcoid;
rdata[0].data = (char *) &xlrec;
rdata[0].len = sizeof(xl_dbase_create_rec);
rdata[0].buffer = InvalidBuffer;
rdata[0].next = NULL;
(void) XLogInsert(RM_DBASE_ID, XLOG_DBASE_CREATE, rdata);
}
/*
* Update the database's pg_database tuple
*/
ScanKeyInit(&scankey,
Anum_pg_database_datname,
BTEqualStrategyNumber, F_NAMEEQ,
NameGetDatum(dbname));
sysscan = systable_beginscan(pgdbrel, DatabaseNameIndexId, true,
SnapshotNow, 1, &scankey);
oldtuple = systable_getnext(sysscan);
if (!HeapTupleIsValid(oldtuple)) /* shouldn't happen... */
ereport(ERROR,
(errcode(ERRCODE_UNDEFINED_DATABASE),
errmsg("database \"%s\" does not exist", dbname)));
MemSet(new_record, 0, sizeof(new_record));
MemSet(new_record_nulls, false, sizeof(new_record_nulls));
MemSet(new_record_repl, false, sizeof(new_record_repl));
new_record[Anum_pg_database_dattablespace - 1] = ObjectIdGetDatum(dst_tblspcoid);
new_record_repl[Anum_pg_database_dattablespace - 1] = true;
newtuple = heap_modify_tuple(oldtuple, RelationGetDescr(pgdbrel),
new_record,
new_record_nulls, new_record_repl);
simple_heap_update(pgdbrel, &oldtuple->t_self, newtuple);
/* Update indexes */
CatalogUpdateIndexes(pgdbrel, newtuple);
systable_endscan(sysscan);
/*
* Force another checkpoint here. As in CREATE DATABASE, this is to
* ensure that we don't have to replay a committed XLOG_DBASE_CREATE
* operation, which would cause us to lose any unlogged operations
* done in the new DB tablespace before the next checkpoint.
*/
RequestCheckpoint(CHECKPOINT_IMMEDIATE | CHECKPOINT_FORCE | CHECKPOINT_WAIT);
/*
* Set flag to update flat database file at commit. Note: this also
* forces synchronous commit, which minimizes the window between
* copying the database files and commital of the transaction. If we
* crash before committing, we'll leave an orphaned set of files on
* disk, which is not fatal but not good either.
*/
database_file_update_needed();
/*
* Close pg_database, but keep lock till commit (this is important to
* prevent any risk of deadlock failure while updating flat file)
*/
heap_close(pgdbrel, NoLock);
}
PG_END_ENSURE_ERROR_CLEANUP(movedb_failure_callback,
PointerGetDatum(&fparms));
/*
* Commit the transaction so that the pg_database update is committed.
* If we crash while removing files, the database won't be corrupt,
* we'll just leave some orphaned files in the old directory.
*
* (This is OK because we know we aren't inside a transaction block.)
*
* XXX would it be safe/better to do this inside the ensure block? Not
* convinced it's a good idea; consider elog just after the transaction
* really commits.
*/
PopActiveSnapshot();
CommitTransactionCommand();
/* Start new transaction for the remaining work; don't need a snapshot */
StartTransactionCommand();
/*
* Remove files from the old tablespace
*/
if (!rmtree(src_dbpath, true))
ereport(WARNING,
(errmsg("some useless files may be left behind in old database directory \"%s\"",
src_dbpath)));
/*
* Record the filesystem change in XLOG
*/
{
xl_dbase_drop_rec xlrec;
XLogRecData rdata[1];
xlrec.db_id = db_id;
xlrec.tablespace_id = src_tblspcoid;
rdata[0].data = (char *) &xlrec;
rdata[0].len = sizeof(xl_dbase_drop_rec);
rdata[0].buffer = InvalidBuffer;
rdata[0].next = NULL;
(void) XLogInsert(RM_DBASE_ID, XLOG_DBASE_DROP, rdata);
}
/* Now it's safe to release the database lock */
UnlockSharedObjectForSession(DatabaseRelationId, db_id, 0,
AccessExclusiveLock);
}
/* Error cleanup callback for movedb */
static void
movedb_failure_callback(int code, Datum arg)
{
movedb_failure_params *fparms = (movedb_failure_params *) DatumGetPointer(arg);
char *dstpath;
/* Get rid of anything we managed to copy to the target directory */
dstpath = GetDatabasePath(fparms->dest_dboid, fparms->dest_tsoid);
(void) rmtree(dstpath, true);
}
/*
* ALTER DATABASE name ...
*/
void
AlterDatabase(AlterDatabaseStmt *stmt)
AlterDatabase(AlterDatabaseStmt *stmt, bool isTopLevel)
{
Relation rel;
HeapTuple tuple,
@ -948,6 +1277,7 @@ AlterDatabase(AlterDatabaseStmt *stmt)
ListCell *option;
int connlimit = -1;
DefElem *dconnlimit = NULL;
DefElem *dtablespace = NULL;
Datum new_record[Natts_pg_database];
bool new_record_nulls[Natts_pg_database];
bool new_record_repl[Natts_pg_database];
@ -965,11 +1295,29 @@ AlterDatabase(AlterDatabaseStmt *stmt)
errmsg("conflicting or redundant options")));
dconnlimit = defel;
}
else if (strcmp(defel->defname, "tablespace") == 0)
{
if (dtablespace)
ereport(ERROR,
(errcode(ERRCODE_SYNTAX_ERROR),
errmsg("conflicting or redundant options")));
dtablespace = defel;
}
else
elog(ERROR, "option \"%s\" not recognized",
defel->defname);
}
if (dtablespace)
{
/* currently, can't be specified along with any other options */
Assert(!dconnlimit);
/* this case isn't allowed within a transaction block */
PreventTransactionChain(isTopLevel, "ALTER DATABASE SET TABLESPACE");
movedb(stmt->dbname, strVal(dtablespace->arg));
return;
}
if (dconnlimit)
connlimit = intVal(dconnlimit->arg);

10
src/backend/parser/gram.y
View File

@ -11,7 +11,7 @@
*
*
* IDENTIFICATION
* $PostgreSQL: pgsql/src/backend/parser/gram.y,v 2.634 2008/10/31 16:36:13 meskes Exp $
* $PostgreSQL: pgsql/src/backend/parser/gram.y,v 2.635 2008/11/07 18:25:06 tgl Exp $
*
* HISTORY
* AUTHOR DATE MAJOR EVENT
@ -5558,6 +5558,14 @@ AlterDatabaseStmt:
n->options = $5;
$$ = (Node *)n;
}
| ALTER DATABASE database_name SET TABLESPACE name
{
AlterDatabaseStmt *n = makeNode(AlterDatabaseStmt);
n->dbname = $3;
n->options = list_make1(makeDefElem("tablespace",
(Node *)makeString($6)));
$$ = (Node *)n;
}
;
AlterDatabaseSetStmt:

41
src/backend/storage/lmgr/lmgr.c
View File

@ -8,7 +8,7 @@
*
*
* IDENTIFICATION
* $PostgreSQL: pgsql/src/backend/storage/lmgr/lmgr.c,v 1.97 2008/03/04 19:54:06 tgl Exp $
* $PostgreSQL: pgsql/src/backend/storage/lmgr/lmgr.c,v 1.98 2008/11/07 18:25:06 tgl Exp $
*
*-------------------------------------------------------------------------
*/
@ -660,6 +660,45 @@ UnlockSharedObject(Oid classid, Oid objid, uint16 objsubid,
LockRelease(&tag, lockmode, false);
}
/*
* LockSharedObjectForSession
*
* Obtain a session-level lock on a shared-across-databases object.
* See LockRelationIdForSession for notes about session-level locks.
*/
void
LockSharedObjectForSession(Oid classid, Oid objid, uint16 objsubid,
LOCKMODE lockmode)
{
LOCKTAG tag;
SET_LOCKTAG_OBJECT(tag,
InvalidOid,
classid,
objid,
objsubid);
(void) LockAcquire(&tag, lockmode, true, false);
}
/*
* UnlockSharedObjectForSession
*/
void
UnlockSharedObjectForSession(Oid classid, Oid objid, uint16 objsubid,
LOCKMODE lockmode)
{
LOCKTAG tag;
SET_LOCKTAG_OBJECT(tag,
InvalidOid,
classid,
objid,
objsubid);
LockRelease(&tag, lockmode, true);
}
/*
* Append a description of a lockable object to buf.

4
src/backend/tcop/utility.c
View File

@ -10,7 +10,7 @@
*
*
* IDENTIFICATION
* $PostgreSQL: pgsql/src/backend/tcop/utility.c,v 1.300 2008/11/04 00:57:19 tgl Exp $
* $PostgreSQL: pgsql/src/backend/tcop/utility.c,v 1.301 2008/11/07 18:25:06 tgl Exp $
*
*-------------------------------------------------------------------------
*/
@ -781,7 +781,7 @@ ProcessUtility(Node *parsetree,
break;
case T_AlterDatabaseStmt:
AlterDatabase((AlterDatabaseStmt *) parsetree);
AlterDatabase((AlterDatabaseStmt *) parsetree, isTopLevel);
break;
case T_AlterDatabaseSetStmt:

9
src/bin/psql/tab-complete.c
View File

@ -3,7 +3,7 @@
*
* Copyright (c) 2000-2008, PostgreSQL Global Development Group
*
* $PostgreSQL: pgsql/src/bin/psql/tab-complete.c,v 1.173 2008/10/04 21:56:54 tgl Exp $
* $PostgreSQL: pgsql/src/bin/psql/tab-complete.c,v 1.174 2008/11/07 18:25:07 tgl Exp $
*/
/*----------------------------------------------------------------------
@ -420,6 +420,7 @@ static const SchemaQuery Query_for_list_of_views = {
" UNION ALL SELECT 'transaction' "\
" UNION ALL SELECT 'session' "\
" UNION ALL SELECT 'role' "\
" UNION ALL SELECT 'tablespace' "\
" UNION ALL SELECT 'all') ss "\
" WHERE substring(name,1,%d)='%s'"
@ -728,7 +729,7 @@ psql_completion(char *text, int start, int end)
pg_strcasecmp(prev2_wd, "INDEX") == 0)
{
static const char *const list_ALTERINDEX[] =
{"SET TABLESPACE", "OWNER TO", "RENAME TO", "SET", "RESET", NULL};
{"OWNER TO", "RENAME TO", "SET", "RESET", NULL};
COMPLETE_WITH_LIST(list_ALTERINDEX);
}
@ -945,9 +946,11 @@ psql_completion(char *text, int start, int end)
COMPLETE_WITH_LIST(list_COLUMNALTER);
}
else if (pg_strcasecmp(prev3_wd, "TABLE") == 0)
else if (pg_strcasecmp(prev3_wd, "TABLE") == 0 &&
pg_strcasecmp(prev_wd, "CLUSTER") == 0)
COMPLETE_WITH_CONST("ON");
else if (pg_strcasecmp(prev4_wd, "TABLE") == 0 &&
pg_strcasecmp(prev2_wd, "CLUSTER") == 0 &&
pg_strcasecmp(prev_wd, "ON") == 0)
{
completion_info_charp = prev3_wd;

4
src/include/commands/dbcommands.h
View File

@ -7,7 +7,7 @@
* Portions Copyright (c) 1996-2008, PostgreSQL Global Development Group
* Portions Copyright (c) 1994, Regents of the University of California
*
* $PostgreSQL: pgsql/src/include/commands/dbcommands.h,v 1.47 2008/01/01 19:45:57 momjian Exp $
* $PostgreSQL: pgsql/src/include/commands/dbcommands.h,v 1.48 2008/11/07 18:25:07 tgl Exp $
*
*-------------------------------------------------------------------------
*/
@ -55,7 +55,7 @@ typedef struct xl_dbase_drop_rec
extern void createdb(const CreatedbStmt *stmt);
extern void dropdb(const char *dbname, bool missing_ok);
extern void RenameDatabase(const char *oldname, const char *newname);
extern void AlterDatabase(AlterDatabaseStmt *stmt);
extern void AlterDatabase(AlterDatabaseStmt *stmt, bool isTopLevel);
extern void AlterDatabaseSet(AlterDatabaseSetStmt *stmt);
extern void AlterDatabaseOwner(const char *dbname, Oid newOwnerId);

7
src/include/storage/lmgr.h
View File

@ -7,7 +7,7 @@
* Portions Copyright (c) 1996-2008, PostgreSQL Global Development Group
* Portions Copyright (c) 1994, Regents of the University of California
*
* $PostgreSQL: pgsql/src/include/storage/lmgr.h,v 1.62 2008/05/12 00:00:53 alvherre Exp $
* $PostgreSQL: pgsql/src/include/storage/lmgr.h,v 1.63 2008/11/07 18:25:07 tgl Exp $
*
*-------------------------------------------------------------------------
*/
@ -73,6 +73,11 @@ extern void LockSharedObject(Oid classid, Oid objid, uint16 objsubid,
extern void UnlockSharedObject(Oid classid, Oid objid, uint16 objsubid,
LOCKMODE lockmode);
extern void LockSharedObjectForSession(Oid classid, Oid objid, uint16 objsubid,
LOCKMODE lockmode);
extern void UnlockSharedObjectForSession(Oid classid, Oid objid, uint16 objsubid,
LOCKMODE lockmode);
/* Describe a locktag for error messages */
extern void DescribeLockTag(StringInfo buf, const LOCKTAG *tag);