1 /*-------------------------------------------------------------------------
4 * Database management commands (create/drop database).
6 * Note: database creation/destruction commands use exclusive locks on
7 * the database objects (as expressed by LockSharedObject()) to avoid
8 * stepping on each others' toes. Formerly we used table-level locks
9 * on pg_database, but that's too coarse-grained.
11 * Portions Copyright (c) 1996-2010, PostgreSQL Global Development Group
12 * Portions Copyright (c) 1994, Regents of the University of California
16 * $PostgreSQL: pgsql/src/backend/commands/dbcommands.c,v 1.230 2010/01/02 16:57:37 momjian Exp $
18 *-------------------------------------------------------------------------
27 #include "access/genam.h"
28 #include "access/heapam.h"
29 #include "access/transam.h"
30 #include "access/xact.h"
31 #include "access/xlogutils.h"
32 #include "catalog/catalog.h"
33 #include "catalog/dependency.h"
34 #include "catalog/indexing.h"
35 #include "catalog/pg_authid.h"
36 #include "catalog/pg_database.h"
37 #include "catalog/pg_db_role_setting.h"
38 #include "catalog/pg_tablespace.h"
39 #include "commands/comment.h"
40 #include "commands/dbcommands.h"
41 #include "commands/tablespace.h"
42 #include "mb/pg_wchar.h"
43 #include "miscadmin.h"
45 #include "postmaster/bgwriter.h"
46 #include "storage/bufmgr.h"
47 #include "storage/fd.h"
48 #include "storage/lmgr.h"
49 #include "storage/ipc.h"
50 #include "storage/procarray.h"
51 #include "storage/smgr.h"
52 #include "storage/standby.h"
53 #include "utils/acl.h"
54 #include "utils/builtins.h"
55 #include "utils/fmgroids.h"
56 #include "utils/lsyscache.h"
57 #include "utils/pg_locale.h"
58 #include "utils/snapmgr.h"
59 #include "utils/syscache.h"
60 #include "utils/tqual.h"
65 Oid src_dboid; /* source (template) DB */
66 Oid dest_dboid; /* DB we are trying to create */
67 } createdb_failure_params;
71 Oid dest_dboid; /* DB we are trying to move */
72 Oid dest_tsoid; /* tablespace we are trying to move to */
73 } movedb_failure_params;
75 /* non-export function prototypes */
76 static void createdb_failure_callback(int code, Datum arg);
77 static void movedb(const char *dbname, const char *tblspcname);
78 static void movedb_failure_callback(int code, Datum arg);
79 static bool get_db_info(const char *name, LOCKMODE lockmode,
80 Oid *dbIdP, Oid *ownerIdP,
81 int *encodingP, bool *dbIsTemplateP, bool *dbAllowConnP,
82 Oid *dbLastSysOidP, TransactionId *dbFrozenXidP,
83 Oid *dbTablespace, char **dbCollate, char **dbCtype);
84 static bool have_createdb_privilege(void);
85 static void remove_dbtablespaces(Oid db_id);
86 static bool check_db_file_conflict(Oid db_id);
87 static int errdetail_busy_db(int notherbackends, int npreparedxacts);
94 createdb(const CreatedbStmt *stmt)
106 TransactionId src_frozenxid;
107 Oid src_deftablespace;
108 volatile Oid dst_deftablespace;
109 Relation pg_database_rel;
111 Datum new_record[Natts_pg_database];
112 bool new_record_nulls[Natts_pg_database];
116 DefElem *dtablespacename = NULL;
117 DefElem *downer = NULL;
118 DefElem *dtemplate = NULL;
119 DefElem *dencoding = NULL;
120 DefElem *dcollate = NULL;
121 DefElem *dctype = NULL;
122 DefElem *dconnlimit = NULL;
123 char *dbname = stmt->dbname;
124 char *dbowner = NULL;
125 const char *dbtemplate = NULL;
126 char *dbcollate = NULL;
127 char *dbctype = NULL;
129 int dbconnlimit = -1;
131 int collate_encoding;
134 createdb_failure_params fparms;
136 /* Extract options from the statement node tree */
137 foreach(option, stmt->options)
139 DefElem *defel = (DefElem *) lfirst(option);
141 if (strcmp(defel->defname, "tablespace") == 0)
145 (errcode(ERRCODE_SYNTAX_ERROR),
146 errmsg("conflicting or redundant options")));
147 dtablespacename = defel;
149 else if (strcmp(defel->defname, "owner") == 0)
153 (errcode(ERRCODE_SYNTAX_ERROR),
154 errmsg("conflicting or redundant options")));
157 else if (strcmp(defel->defname, "template") == 0)
161 (errcode(ERRCODE_SYNTAX_ERROR),
162 errmsg("conflicting or redundant options")));
165 else if (strcmp(defel->defname, "encoding") == 0)
169 (errcode(ERRCODE_SYNTAX_ERROR),
170 errmsg("conflicting or redundant options")));
173 else if (strcmp(defel->defname, "lc_collate") == 0)
177 (errcode(ERRCODE_SYNTAX_ERROR),
178 errmsg("conflicting or redundant options")));
181 else if (strcmp(defel->defname, "lc_ctype") == 0)
185 (errcode(ERRCODE_SYNTAX_ERROR),
186 errmsg("conflicting or redundant options")));
189 else if (strcmp(defel->defname, "connectionlimit") == 0)
193 (errcode(ERRCODE_SYNTAX_ERROR),
194 errmsg("conflicting or redundant options")));
197 else if (strcmp(defel->defname, "location") == 0)
200 (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
201 errmsg("LOCATION is not supported anymore"),
202 errhint("Consider using tablespaces instead.")));
205 elog(ERROR, "option \"%s\" not recognized",
209 if (downer && downer->arg)
210 dbowner = strVal(downer->arg);
211 if (dtemplate && dtemplate->arg)
212 dbtemplate = strVal(dtemplate->arg);
213 if (dencoding && dencoding->arg)
215 const char *encoding_name;
217 if (IsA(dencoding->arg, Integer))
219 encoding = intVal(dencoding->arg);
220 encoding_name = pg_encoding_to_char(encoding);
221 if (strcmp(encoding_name, "") == 0 ||
222 pg_valid_server_encoding(encoding_name) < 0)
224 (errcode(ERRCODE_UNDEFINED_OBJECT),
225 errmsg("%d is not a valid encoding code",
228 else if (IsA(dencoding->arg, String))
230 encoding_name = strVal(dencoding->arg);
231 encoding = pg_valid_server_encoding(encoding_name);
234 (errcode(ERRCODE_UNDEFINED_OBJECT),
235 errmsg("%s is not a valid encoding name",
239 elog(ERROR, "unrecognized node type: %d",
240 nodeTag(dencoding->arg));
242 if (dcollate && dcollate->arg)
243 dbcollate = strVal(dcollate->arg);
244 if (dctype && dctype->arg)
245 dbctype = strVal(dctype->arg);
247 if (dconnlimit && dconnlimit->arg)
249 dbconnlimit = intVal(dconnlimit->arg);
250 if (dbconnlimit < -1)
252 (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
253 errmsg("invalid connection limit: %d", dbconnlimit)));
256 /* obtain OID of proposed owner */
258 datdba = get_roleid_checked(dbowner);
260 datdba = GetUserId();
263 * To create a database, must have createdb privilege and must be able to
264 * become the target role (this does not imply that the target role itself
265 * must have createdb privilege). The latter provision guards against
266 * "giveaway" attacks. Note that a superuser will always have both of
267 * these privileges a fortiori.
269 if (!have_createdb_privilege())
271 (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
272 errmsg("permission denied to create database")));
274 check_is_member_of_role(GetUserId(), datdba);
277 * Lookup database (template) to be cloned, and obtain share lock on it.
278 * ShareLock allows two CREATE DATABASEs to work from the same template
279 * concurrently, while ensuring no one is busy dropping it in parallel
280 * (which would be Very Bad since we'd likely get an incomplete copy
281 * without knowing it). This also prevents any new connections from being
282 * made to the source until we finish copying it, so we can be sure it
283 * won't change underneath us.
286 dbtemplate = "template1"; /* Default template database name */
288 if (!get_db_info(dbtemplate, ShareLock,
289 &src_dboid, &src_owner, &src_encoding,
290 &src_istemplate, &src_allowconn, &src_lastsysoid,
291 &src_frozenxid, &src_deftablespace,
292 &src_collate, &src_ctype))
294 (errcode(ERRCODE_UNDEFINED_DATABASE),
295 errmsg("template database \"%s\" does not exist",
299 * Permission check: to copy a DB that's not marked datistemplate, you
300 * must be superuser or the owner thereof.
304 if (!pg_database_ownercheck(src_dboid, GetUserId()))
306 (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
307 errmsg("permission denied to copy database \"%s\"",
311 /* If encoding or locales are defaulted, use source's setting */
313 encoding = src_encoding;
314 if (dbcollate == NULL)
315 dbcollate = src_collate;
319 /* Some encodings are client only */
320 if (!PG_VALID_BE_ENCODING(encoding))
322 (errcode(ERRCODE_WRONG_OBJECT_TYPE),
323 errmsg("invalid server encoding %d", encoding)));
325 /* Check that the chosen locales are valid */
326 if (!check_locale(LC_COLLATE, dbcollate))
328 (errcode(ERRCODE_WRONG_OBJECT_TYPE),
329 errmsg("invalid locale name %s", dbcollate)));
330 if (!check_locale(LC_CTYPE, dbctype))
332 (errcode(ERRCODE_WRONG_OBJECT_TYPE),
333 errmsg("invalid locale name %s", dbctype)));
336 * Check whether chosen encoding matches chosen locale settings. This
337 * restriction is necessary because libc's locale-specific code usually
338 * fails when presented with data in an encoding it's not expecting. We
339 * allow mismatch in four cases:
341 * 1. locale encoding = SQL_ASCII, which means that the locale is
342 * C/POSIX which works with any encoding.
344 * 2. locale encoding = -1, which means that we couldn't determine
345 * the locale's encoding and have to trust the user to get it right.
347 * 3. selected encoding is UTF8 and platform is win32. This is because
348 * UTF8 is a pseudo codepage that is supported in all locales since it's
349 * converted to UTF16 before being used.
351 * 4. selected encoding is SQL_ASCII, but only if you're a superuser. This
352 * is risky but we have historically allowed it --- notably, the
353 * regression tests require it.
355 * Note: if you change this policy, fix initdb to match.
357 ctype_encoding = pg_get_encoding_from_locale(dbctype);
358 collate_encoding = pg_get_encoding_from_locale(dbcollate);
360 if (!(ctype_encoding == encoding ||
361 ctype_encoding == PG_SQL_ASCII ||
362 ctype_encoding == -1 ||
364 encoding == PG_UTF8 ||
366 (encoding == PG_SQL_ASCII && superuser())))
368 (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
369 errmsg("encoding %s does not match locale %s",
370 pg_encoding_to_char(encoding),
372 errdetail("The chosen LC_CTYPE setting requires encoding %s.",
373 pg_encoding_to_char(ctype_encoding))));
375 if (!(collate_encoding == encoding ||
376 collate_encoding == PG_SQL_ASCII ||
377 collate_encoding == -1 ||
379 encoding == PG_UTF8 ||
381 (encoding == PG_SQL_ASCII && superuser())))
383 (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
384 errmsg("encoding %s does not match locale %s",
385 pg_encoding_to_char(encoding),
387 errdetail("The chosen LC_COLLATE setting requires encoding %s.",
388 pg_encoding_to_char(collate_encoding))));
391 * Check that the new encoding and locale settings match the source
392 * database. We insist on this because we simply copy the source data ---
393 * any non-ASCII data would be wrongly encoded, and any indexes sorted
394 * according to the source locale would be wrong.
396 * However, we assume that template0 doesn't contain any non-ASCII data
397 * nor any indexes that depend on collation or ctype, so template0 can be
398 * used as template for creating a database with any encoding or locale.
400 if (strcmp(dbtemplate, "template0") != 0)
402 if (encoding != src_encoding)
404 (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
405 errmsg("new encoding (%s) is incompatible with the encoding of the template database (%s)",
406 pg_encoding_to_char(encoding),
407 pg_encoding_to_char(src_encoding)),
408 errhint("Use the same encoding as in the template database, or use template0 as template.")));
410 if (strcmp(dbcollate, src_collate) != 0)
412 (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
413 errmsg("new collation (%s) is incompatible with the collation of the template database (%s)",
414 dbcollate, src_collate),
415 errhint("Use the same collation as in the template database, or use template0 as template.")));
417 if (strcmp(dbctype, src_ctype) != 0)
419 (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
420 errmsg("new LC_CTYPE (%s) is incompatible with the LC_CTYPE of the template database (%s)",
422 errhint("Use the same LC_CTYPE as in the template database, or use template0 as template.")));
425 /* Resolve default tablespace for new database */
426 if (dtablespacename && dtablespacename->arg)
428 char *tablespacename;
431 tablespacename = strVal(dtablespacename->arg);
432 dst_deftablespace = get_tablespace_oid(tablespacename);
433 if (!OidIsValid(dst_deftablespace))
435 (errcode(ERRCODE_UNDEFINED_OBJECT),
436 errmsg("tablespace \"%s\" does not exist",
438 /* check permissions */
439 aclresult = pg_tablespace_aclcheck(dst_deftablespace, GetUserId(),
441 if (aclresult != ACLCHECK_OK)
442 aclcheck_error(aclresult, ACL_KIND_TABLESPACE,
445 /* pg_global must never be the default tablespace */
446 if (dst_deftablespace == GLOBALTABLESPACE_OID)
448 (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
449 errmsg("pg_global cannot be used as default tablespace")));
452 * If we are trying to change the default tablespace of the template,
453 * we require that the template not have any files in the new default
454 * tablespace. This is necessary because otherwise the copied
455 * database would contain pg_class rows that refer to its default
456 * tablespace both explicitly (by OID) and implicitly (as zero), which
457 * would cause problems. For example another CREATE DATABASE using
458 * the copied database as template, and trying to change its default
459 * tablespace again, would yield outright incorrect results (it would
460 * improperly move tables to the new default tablespace that should
461 * stay in the same tablespace).
463 if (dst_deftablespace != src_deftablespace)
468 srcpath = GetDatabasePath(src_dboid, dst_deftablespace);
470 if (stat(srcpath, &st) == 0 &&
471 S_ISDIR(st.st_mode) &&
472 !directory_is_empty(srcpath))
474 (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
475 errmsg("cannot assign new default tablespace \"%s\"",
477 errdetail("There is a conflict because database \"%s\" already has some tables in this tablespace.",
484 /* Use template database's default tablespace */
485 dst_deftablespace = src_deftablespace;
486 /* Note there is no additional permission check in this path */
490 * Check for db name conflict. This is just to give a more friendly error
491 * message than "unique index violation". There's a race condition but
492 * we're willing to accept the less friendly message in that case.
494 if (OidIsValid(get_database_oid(dbname)))
496 (errcode(ERRCODE_DUPLICATE_DATABASE),
497 errmsg("database \"%s\" already exists", dbname)));
500 * The source DB can't have any active backends, except this one
501 * (exception is to allow CREATE DB while connected to template1).
502 * Otherwise we might copy inconsistent data.
504 * This should be last among the basic error checks, because it involves
505 * potential waiting; we may as well throw an error first if we're gonna
508 if (CountOtherDBBackends(src_dboid, ¬herbackends, &npreparedxacts))
510 (errcode(ERRCODE_OBJECT_IN_USE),
511 errmsg("source database \"%s\" is being accessed by other users",
513 errdetail_busy_db(notherbackends, npreparedxacts)));
516 * Select an OID for the new database, checking that it doesn't have a
517 * filename conflict with anything already existing in the tablespace
520 pg_database_rel = heap_open(DatabaseRelationId, RowExclusiveLock);
524 dboid = GetNewOid(pg_database_rel);
525 } while (check_db_file_conflict(dboid));
528 * Insert a new tuple into pg_database. This establishes our ownership of
529 * the new database name (anyone else trying to insert the same name will
530 * block on the unique index, and fail after we commit).
534 MemSet(new_record, 0, sizeof(new_record));
535 MemSet(new_record_nulls, false, sizeof(new_record_nulls));
537 new_record[Anum_pg_database_datname - 1] =
538 DirectFunctionCall1(namein, CStringGetDatum(dbname));
539 new_record[Anum_pg_database_datdba - 1] = ObjectIdGetDatum(datdba);
540 new_record[Anum_pg_database_encoding - 1] = Int32GetDatum(encoding);
541 new_record[Anum_pg_database_datcollate - 1] =
542 DirectFunctionCall1(namein, CStringGetDatum(dbcollate));
543 new_record[Anum_pg_database_datctype - 1] =
544 DirectFunctionCall1(namein, CStringGetDatum(dbctype));
545 new_record[Anum_pg_database_datistemplate - 1] = BoolGetDatum(false);
546 new_record[Anum_pg_database_datallowconn - 1] = BoolGetDatum(true);
547 new_record[Anum_pg_database_datconnlimit - 1] = Int32GetDatum(dbconnlimit);
548 new_record[Anum_pg_database_datlastsysoid - 1] = ObjectIdGetDatum(src_lastsysoid);
549 new_record[Anum_pg_database_datfrozenxid - 1] = TransactionIdGetDatum(src_frozenxid);
550 new_record[Anum_pg_database_dattablespace - 1] = ObjectIdGetDatum(dst_deftablespace);
553 * We deliberately set datacl to default (NULL), rather than copying it
554 * from the template database. Copying it would be a bad idea when the
555 * owner is not the same as the template's owner.
557 new_record_nulls[Anum_pg_database_datacl - 1] = true;
559 tuple = heap_form_tuple(RelationGetDescr(pg_database_rel),
560 new_record, new_record_nulls);
562 HeapTupleSetOid(tuple, dboid);
564 simple_heap_insert(pg_database_rel, tuple);
567 CatalogUpdateIndexes(pg_database_rel, tuple);
570 * Now generate additional catalog entries associated with the new DB
573 /* Register owner dependency */
574 recordDependencyOnOwner(DatabaseRelationId, dboid, datdba);
576 /* Create pg_shdepend entries for objects within database */
577 copyTemplateDependencies(src_dboid, dboid);
580 * Force a checkpoint before starting the copy. This will force dirty
581 * buffers out to disk, to ensure source database is up-to-date on disk
582 * for the copy. FlushDatabaseBuffers() would suffice for that, but we
583 * also want to process any pending unlink requests. Otherwise, if a
584 * checkpoint happened while we're copying files, a file might be deleted
585 * just when we're about to copy it, causing the lstat() call in copydir()
586 * to fail with ENOENT.
588 RequestCheckpoint(CHECKPOINT_IMMEDIATE | CHECKPOINT_FORCE | CHECKPOINT_WAIT);
591 * Once we start copying subdirectories, we need to be able to clean 'em
592 * up if we fail. Use an ENSURE block to make sure this happens. (This
593 * is not a 100% solution, because of the possibility of failure during
594 * transaction commit after we leave this routine, but it should handle
597 fparms.src_dboid = src_dboid;
598 fparms.dest_dboid = dboid;
599 PG_ENSURE_ERROR_CLEANUP(createdb_failure_callback,
600 PointerGetDatum(&fparms));
603 * Iterate through all tablespaces of the template database, and copy
604 * each one to the new database.
606 rel = heap_open(TableSpaceRelationId, AccessShareLock);
607 scan = heap_beginscan(rel, SnapshotNow, 0, NULL);
608 while ((tuple = heap_getnext(scan, ForwardScanDirection)) != NULL)
610 Oid srctablespace = HeapTupleGetOid(tuple);
616 /* No need to copy global tablespace */
617 if (srctablespace == GLOBALTABLESPACE_OID)
620 srcpath = GetDatabasePath(src_dboid, srctablespace);
622 if (stat(srcpath, &st) < 0 || !S_ISDIR(st.st_mode) ||
623 directory_is_empty(srcpath))
625 /* Assume we can ignore it */
630 if (srctablespace == src_deftablespace)
631 dsttablespace = dst_deftablespace;
633 dsttablespace = srctablespace;
635 dstpath = GetDatabasePath(dboid, dsttablespace);
638 * Copy this subdirectory to the new location
640 * We don't need to copy subdirectories
642 copydir(srcpath, dstpath, false);
644 /* Record the filesystem change in XLOG */
646 xl_dbase_create_rec xlrec;
647 XLogRecData rdata[1];
650 xlrec.tablespace_id = dsttablespace;
651 xlrec.src_db_id = src_dboid;
652 xlrec.src_tablespace_id = srctablespace;
654 rdata[0].data = (char *) &xlrec;
655 rdata[0].len = sizeof(xl_dbase_create_rec);
656 rdata[0].buffer = InvalidBuffer;
657 rdata[0].next = NULL;
659 (void) XLogInsert(RM_DBASE_ID, XLOG_DBASE_CREATE, rdata);
663 heap_close(rel, AccessShareLock);
666 * We force a checkpoint before committing. This effectively means
667 * that committed XLOG_DBASE_CREATE operations will never need to be
668 * replayed (at least not in ordinary crash recovery; we still have to
669 * make the XLOG entry for the benefit of PITR operations). This
670 * avoids two nasty scenarios:
672 * #1: When PITR is off, we don't XLOG the contents of newly created
673 * indexes; therefore the drop-and-recreate-whole-directory behavior
674 * of DBASE_CREATE replay would lose such indexes.
676 * #2: Since we have to recopy the source database during DBASE_CREATE
677 * replay, we run the risk of copying changes in it that were
678 * committed after the original CREATE DATABASE command but before the
679 * system crash that led to the replay. This is at least unexpected
680 * and at worst could lead to inconsistencies, eg duplicate table
683 * (Both of these were real bugs in releases 8.0 through 8.0.3.)
685 * In PITR replay, the first of these isn't an issue, and the second
686 * is only a risk if the CREATE DATABASE and subsequent template
687 * database change both occur while a base backup is being taken.
688 * There doesn't seem to be much we can do about that except document
689 * it as a limitation.
691 * Perhaps if we ever implement CREATE DATABASE in a less cheesy way,
694 RequestCheckpoint(CHECKPOINT_IMMEDIATE | CHECKPOINT_FORCE | CHECKPOINT_WAIT);
697 * Close pg_database, but keep lock till commit.
699 heap_close(pg_database_rel, NoLock);
702 * Force synchronous commit, thus minimizing the window between
703 * creation of the database files and commital of the transaction. If
704 * we crash before committing, we'll have a DB that's taking up disk
705 * space but is not in pg_database, which is not good.
709 PG_END_ENSURE_ERROR_CLEANUP(createdb_failure_callback,
710 PointerGetDatum(&fparms));
713 /* Error cleanup callback for createdb */
715 createdb_failure_callback(int code, Datum arg)
717 createdb_failure_params *fparms = (createdb_failure_params *) DatumGetPointer(arg);
720 * Release lock on source database before doing recursive remove. This is
721 * not essential but it seems desirable to release the lock as soon as
724 UnlockSharedObject(DatabaseRelationId, fparms->src_dboid, 0, ShareLock);
726 /* Throw away any successfully copied subdirectories */
727 remove_dbtablespaces(fparms->dest_dboid);
735 dropdb(const char *dbname, bool missing_ok)
745 * Look up the target database's OID, and get exclusive lock on it. We
746 * need this to ensure that no new backend starts up in the target
747 * database while we are deleting it (see postinit.c), and that no one is
748 * using it as a CREATE DATABASE template or trying to delete it for
751 pgdbrel = heap_open(DatabaseRelationId, RowExclusiveLock);
753 if (!get_db_info(dbname, AccessExclusiveLock, &db_id, NULL, NULL,
754 &db_istemplate, NULL, NULL, NULL, NULL, NULL, NULL))
759 (errcode(ERRCODE_UNDEFINED_DATABASE),
760 errmsg("database \"%s\" does not exist", dbname)));
764 /* Close pg_database, release the lock, since we changed nothing */
765 heap_close(pgdbrel, RowExclusiveLock);
767 (errmsg("database \"%s\" does not exist, skipping",
776 if (!pg_database_ownercheck(db_id, GetUserId()))
777 aclcheck_error(ACLCHECK_NOT_OWNER, ACL_KIND_DATABASE,
781 * Disallow dropping a DB that is marked istemplate. This is just to
782 * prevent people from accidentally dropping template0 or template1; they
783 * can do so if they're really determined ...
787 (errcode(ERRCODE_WRONG_OBJECT_TYPE),
788 errmsg("cannot drop a template database")));
790 /* Obviously can't drop my own database */
791 if (db_id == MyDatabaseId)
793 (errcode(ERRCODE_OBJECT_IN_USE),
794 errmsg("cannot drop the currently open database")));
797 * Check for other backends in the target database. (Because we hold the
798 * database lock, no new ones can start after this.)
800 * As in CREATE DATABASE, check this after other error conditions.
802 if (CountOtherDBBackends(db_id, ¬herbackends, &npreparedxacts))
804 (errcode(ERRCODE_OBJECT_IN_USE),
805 errmsg("database \"%s\" is being accessed by other users",
807 errdetail_busy_db(notherbackends, npreparedxacts)));
810 * Remove the database's tuple from pg_database.
812 tup = SearchSysCache(DATABASEOID,
813 ObjectIdGetDatum(db_id),
815 if (!HeapTupleIsValid(tup))
816 elog(ERROR, "cache lookup failed for database %u", db_id);
818 simple_heap_delete(pgdbrel, &tup->t_self);
820 ReleaseSysCache(tup);
823 * Delete any comments associated with the database.
825 DeleteSharedComments(db_id, DatabaseRelationId);
828 * Remove settings associated with this database
830 DropSetting(db_id, InvalidOid);
833 * Remove shared dependency references for the database.
835 dropDatabaseDependencies(db_id);
838 * Drop pages for this database that are in the shared buffer cache. This
839 * is important to ensure that no remaining backend tries to write out a
840 * dirty buffer to the dead database later...
842 DropDatabaseBuffers(db_id);
845 * Tell the stats collector to forget it immediately, too.
847 pgstat_drop_database(db_id);
850 * Tell bgwriter to forget any pending fsync and unlink requests for files
851 * in the database; else the fsyncs will fail at next checkpoint, or
852 * worse, it will delete files that belong to a newly created database
855 ForgetDatabaseFsyncRequests(db_id);
858 * Force a checkpoint to make sure the bgwriter has received the message
859 * sent by ForgetDatabaseFsyncRequests. On Windows, this also ensures that
860 * the bgwriter doesn't hold any open files, which would cause rmdir() to
863 RequestCheckpoint(CHECKPOINT_IMMEDIATE | CHECKPOINT_FORCE | CHECKPOINT_WAIT);
866 * Remove all tablespace subdirs belonging to the database.
868 remove_dbtablespaces(db_id);
871 * Close pg_database, but keep lock till commit.
873 heap_close(pgdbrel, NoLock);
876 * Force synchronous commit, thus minimizing the window between removal
877 * of the database files and commital of the transaction. If we crash
878 * before committing, we'll have a DB that's gone on disk but still there
879 * according to pg_database, which is not good.
889 RenameDatabase(const char *oldname, const char *newname)
898 * Look up the target database's OID, and get exclusive lock on it. We
899 * need this for the same reasons as DROP DATABASE.
901 rel = heap_open(DatabaseRelationId, RowExclusiveLock);
903 if (!get_db_info(oldname, AccessExclusiveLock, &db_id, NULL, NULL,
904 NULL, NULL, NULL, NULL, NULL, NULL, NULL))
906 (errcode(ERRCODE_UNDEFINED_DATABASE),
907 errmsg("database \"%s\" does not exist", oldname)));
910 if (!pg_database_ownercheck(db_id, GetUserId()))
911 aclcheck_error(ACLCHECK_NOT_OWNER, ACL_KIND_DATABASE,
914 /* must have createdb rights */
915 if (!have_createdb_privilege())
917 (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
918 errmsg("permission denied to rename database")));
921 * Make sure the new name doesn't exist. See notes for same error in
924 if (OidIsValid(get_database_oid(newname)))
926 (errcode(ERRCODE_DUPLICATE_DATABASE),
927 errmsg("database \"%s\" already exists", newname)));
930 * XXX Client applications probably store the current database somewhere,
931 * so renaming it could cause confusion. On the other hand, there may not
932 * be an actual problem besides a little confusion, so think about this
935 if (db_id == MyDatabaseId)
937 (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
938 errmsg("current database cannot be renamed")));
941 * Make sure the database does not have active sessions. This is the same
942 * concern as above, but applied to other sessions.
944 * As in CREATE DATABASE, check this after other error conditions.
946 if (CountOtherDBBackends(db_id, ¬herbackends, &npreparedxacts))
948 (errcode(ERRCODE_OBJECT_IN_USE),
949 errmsg("database \"%s\" is being accessed by other users",
951 errdetail_busy_db(notherbackends, npreparedxacts)));
954 newtup = SearchSysCacheCopy(DATABASEOID,
955 ObjectIdGetDatum(db_id),
957 if (!HeapTupleIsValid(newtup))
958 elog(ERROR, "cache lookup failed for database %u", db_id);
959 namestrcpy(&(((Form_pg_database) GETSTRUCT(newtup))->datname), newname);
960 simple_heap_update(rel, &newtup->t_self, newtup);
961 CatalogUpdateIndexes(rel, newtup);
964 * Close pg_database, but keep lock till commit.
966 heap_close(rel, NoLock);
971 * ALTER DATABASE SET TABLESPACE
974 movedb(const char *dbname, const char *tblspcname)
984 Datum new_record[Natts_pg_database];
985 bool new_record_nulls[Natts_pg_database];
986 bool new_record_repl[Natts_pg_database];
994 movedb_failure_params fparms;
997 * Look up the target database's OID, and get exclusive lock on it. We
998 * need this to ensure that no new backend starts up in the database while
999 * we are moving it, and that no one is using it as a CREATE DATABASE
1000 * template or trying to delete it.
1002 pgdbrel = heap_open(DatabaseRelationId, RowExclusiveLock);
1004 if (!get_db_info(dbname, AccessExclusiveLock, &db_id, NULL, NULL,
1005 NULL, NULL, NULL, NULL, &src_tblspcoid, NULL, NULL))
1007 (errcode(ERRCODE_UNDEFINED_DATABASE),
1008 errmsg("database \"%s\" does not exist", dbname)));
1011 * We actually need a session lock, so that the lock will persist across
1012 * the commit/restart below. (We could almost get away with letting the
1013 * lock be released at commit, except that someone could try to move
1014 * relations of the DB back into the old directory while we rmtree() it.)
1016 LockSharedObjectForSession(DatabaseRelationId, db_id, 0,
1017 AccessExclusiveLock);
1022 if (!pg_database_ownercheck(db_id, GetUserId()))
1023 aclcheck_error(ACLCHECK_NOT_OWNER, ACL_KIND_DATABASE,
1027 * Obviously can't move the tables of my own database
1029 if (db_id == MyDatabaseId)
1031 (errcode(ERRCODE_OBJECT_IN_USE),
1032 errmsg("cannot change the tablespace of the currently open database")));
1035 * Get tablespace's oid
1037 dst_tblspcoid = get_tablespace_oid(tblspcname);
1038 if (dst_tblspcoid == InvalidOid)
1040 (errcode(ERRCODE_UNDEFINED_DATABASE),
1041 errmsg("tablespace \"%s\" does not exist", tblspcname)));
1046 aclresult = pg_tablespace_aclcheck(dst_tblspcoid, GetUserId(),
1048 if (aclresult != ACLCHECK_OK)
1049 aclcheck_error(aclresult, ACL_KIND_TABLESPACE,
1053 * pg_global must never be the default tablespace
1055 if (dst_tblspcoid == GLOBALTABLESPACE_OID)
1057 (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
1058 errmsg("pg_global cannot be used as default tablespace")));
1061 * No-op if same tablespace
1063 if (src_tblspcoid == dst_tblspcoid)
1065 heap_close(pgdbrel, NoLock);
1066 UnlockSharedObjectForSession(DatabaseRelationId, db_id, 0,
1067 AccessExclusiveLock);
1072 * Check for other backends in the target database. (Because we hold the
1073 * database lock, no new ones can start after this.)
1075 * As in CREATE DATABASE, check this after other error conditions.
1077 if (CountOtherDBBackends(db_id, ¬herbackends, &npreparedxacts))
1079 (errcode(ERRCODE_OBJECT_IN_USE),
1080 errmsg("database \"%s\" is being accessed by other users",
1082 errdetail_busy_db(notherbackends, npreparedxacts)));
1085 * Get old and new database paths
1087 src_dbpath = GetDatabasePath(db_id, src_tblspcoid);
1088 dst_dbpath = GetDatabasePath(db_id, dst_tblspcoid);
1091 * Force a checkpoint before proceeding. This will force dirty buffers out
1092 * to disk, to ensure source database is up-to-date on disk for the copy.
1093 * FlushDatabaseBuffers() would suffice for that, but we also want to
1094 * process any pending unlink requests. Otherwise, the check for existing
1095 * files in the target directory might fail unnecessarily, not to mention
1096 * that the copy might fail due to source files getting deleted under it.
1097 * On Windows, this also ensures that the bgwriter doesn't hold any open
1098 * files, which would cause rmdir() to fail.
1100 RequestCheckpoint(CHECKPOINT_IMMEDIATE | CHECKPOINT_FORCE | CHECKPOINT_WAIT);
1103 * Check for existence of files in the target directory, i.e., objects of
1104 * this database that are already in the target tablespace. We can't
1105 * allow the move in such a case, because we would need to change those
1106 * relations' pg_class.reltablespace entries to zero, and we don't have
1107 * access to the DB's pg_class to do so.
1109 dstdir = AllocateDir(dst_dbpath);
1112 while ((xlde = ReadDir(dstdir, dst_dbpath)) != NULL)
1114 if (strcmp(xlde->d_name, ".") == 0 ||
1115 strcmp(xlde->d_name, "..") == 0)
1119 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1120 errmsg("some relations of database \"%s\" are already in tablespace \"%s\"",
1121 dbname, tblspcname),
1122 errhint("You must move them back to the database's default tablespace before using this command.")));
1128 * The directory exists but is empty. We must remove it before using
1129 * the copydir function.
1131 if (rmdir(dst_dbpath) != 0)
1132 elog(ERROR, "could not remove directory \"%s\": %m",
1137 * Use an ENSURE block to make sure we remove the debris if the copy fails
1138 * (eg, due to out-of-disk-space). This is not a 100% solution, because
1139 * of the possibility of failure during transaction commit, but it should
1140 * handle most scenarios.
1142 fparms.dest_dboid = db_id;
1143 fparms.dest_tsoid = dst_tblspcoid;
1144 PG_ENSURE_ERROR_CLEANUP(movedb_failure_callback,
1145 PointerGetDatum(&fparms));
1148 * Copy files from the old tablespace to the new one
1150 copydir(src_dbpath, dst_dbpath, false);
1153 * Record the filesystem change in XLOG
1156 xl_dbase_create_rec xlrec;
1157 XLogRecData rdata[1];
1159 xlrec.db_id = db_id;
1160 xlrec.tablespace_id = dst_tblspcoid;
1161 xlrec.src_db_id = db_id;
1162 xlrec.src_tablespace_id = src_tblspcoid;
1164 rdata[0].data = (char *) &xlrec;
1165 rdata[0].len = sizeof(xl_dbase_create_rec);
1166 rdata[0].buffer = InvalidBuffer;
1167 rdata[0].next = NULL;
1169 (void) XLogInsert(RM_DBASE_ID, XLOG_DBASE_CREATE, rdata);
1173 * Update the database's pg_database tuple
1175 ScanKeyInit(&scankey,
1176 Anum_pg_database_datname,
1177 BTEqualStrategyNumber, F_NAMEEQ,
1178 NameGetDatum(dbname));
1179 sysscan = systable_beginscan(pgdbrel, DatabaseNameIndexId, true,
1180 SnapshotNow, 1, &scankey);
1181 oldtuple = systable_getnext(sysscan);
1182 if (!HeapTupleIsValid(oldtuple)) /* shouldn't happen... */
1184 (errcode(ERRCODE_UNDEFINED_DATABASE),
1185 errmsg("database \"%s\" does not exist", dbname)));
1187 MemSet(new_record, 0, sizeof(new_record));
1188 MemSet(new_record_nulls, false, sizeof(new_record_nulls));
1189 MemSet(new_record_repl, false, sizeof(new_record_repl));
1191 new_record[Anum_pg_database_dattablespace - 1] = ObjectIdGetDatum(dst_tblspcoid);
1192 new_record_repl[Anum_pg_database_dattablespace - 1] = true;
1194 newtuple = heap_modify_tuple(oldtuple, RelationGetDescr(pgdbrel),
1196 new_record_nulls, new_record_repl);
1197 simple_heap_update(pgdbrel, &oldtuple->t_self, newtuple);
1199 /* Update indexes */
1200 CatalogUpdateIndexes(pgdbrel, newtuple);
1202 systable_endscan(sysscan);
1205 * Force another checkpoint here. As in CREATE DATABASE, this is to
1206 * ensure that we don't have to replay a committed XLOG_DBASE_CREATE
1207 * operation, which would cause us to lose any unlogged operations
1208 * done in the new DB tablespace before the next checkpoint.
1210 RequestCheckpoint(CHECKPOINT_IMMEDIATE | CHECKPOINT_FORCE | CHECKPOINT_WAIT);
1213 * Force synchronous commit, thus minimizing the window between
1214 * copying the database files and commital of the transaction. If we
1215 * crash before committing, we'll leave an orphaned set of files on
1216 * disk, which is not fatal but not good either.
1221 * Close pg_database, but keep lock till commit.
1223 heap_close(pgdbrel, NoLock);
1225 PG_END_ENSURE_ERROR_CLEANUP(movedb_failure_callback,
1226 PointerGetDatum(&fparms));
1229 * Commit the transaction so that the pg_database update is committed. If
1230 * we crash while removing files, the database won't be corrupt, we'll
1231 * just leave some orphaned files in the old directory.
1233 * (This is OK because we know we aren't inside a transaction block.)
1235 * XXX would it be safe/better to do this inside the ensure block? Not
1236 * convinced it's a good idea; consider elog just after the transaction
1239 PopActiveSnapshot();
1240 CommitTransactionCommand();
1242 /* Start new transaction for the remaining work; don't need a snapshot */
1243 StartTransactionCommand();
1246 * Remove files from the old tablespace
1248 if (!rmtree(src_dbpath, true))
1250 (errmsg("some useless files may be left behind in old database directory \"%s\"",
1254 * Record the filesystem change in XLOG
1257 xl_dbase_drop_rec xlrec;
1258 XLogRecData rdata[1];
1260 xlrec.db_id = db_id;
1261 xlrec.tablespace_id = src_tblspcoid;
1263 rdata[0].data = (char *) &xlrec;
1264 rdata[0].len = sizeof(xl_dbase_drop_rec);
1265 rdata[0].buffer = InvalidBuffer;
1266 rdata[0].next = NULL;
1268 (void) XLogInsert(RM_DBASE_ID, XLOG_DBASE_DROP, rdata);
1271 /* Now it's safe to release the database lock */
1272 UnlockSharedObjectForSession(DatabaseRelationId, db_id, 0,
1273 AccessExclusiveLock);
1276 /* Error cleanup callback for movedb */
1278 movedb_failure_callback(int code, Datum arg)
1280 movedb_failure_params *fparms = (movedb_failure_params *) DatumGetPointer(arg);
1283 /* Get rid of anything we managed to copy to the target directory */
1284 dstpath = GetDatabasePath(fparms->dest_dboid, fparms->dest_tsoid);
1286 (void) rmtree(dstpath, true);
1291 * ALTER DATABASE name ...
1294 AlterDatabase(AlterDatabaseStmt *stmt, bool isTopLevel)
1299 ScanKeyData scankey;
1303 DefElem *dconnlimit = NULL;
1304 DefElem *dtablespace = NULL;
1305 Datum new_record[Natts_pg_database];
1306 bool new_record_nulls[Natts_pg_database];
1307 bool new_record_repl[Natts_pg_database];
1309 /* Extract options from the statement node tree */
1310 foreach(option, stmt->options)
1312 DefElem *defel = (DefElem *) lfirst(option);
1314 if (strcmp(defel->defname, "connectionlimit") == 0)
1318 (errcode(ERRCODE_SYNTAX_ERROR),
1319 errmsg("conflicting or redundant options")));
1322 else if (strcmp(defel->defname, "tablespace") == 0)
1326 (errcode(ERRCODE_SYNTAX_ERROR),
1327 errmsg("conflicting or redundant options")));
1328 dtablespace = defel;
1331 elog(ERROR, "option \"%s\" not recognized",
1337 /* currently, can't be specified along with any other options */
1338 Assert(!dconnlimit);
1339 /* this case isn't allowed within a transaction block */
1340 PreventTransactionChain(isTopLevel, "ALTER DATABASE SET TABLESPACE");
1341 movedb(stmt->dbname, strVal(dtablespace->arg));
1347 connlimit = intVal(dconnlimit->arg);
1350 (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
1351 errmsg("invalid connection limit: %d", connlimit)));
1355 * Get the old tuple. We don't need a lock on the database per se,
1356 * because we're not going to do anything that would mess up incoming
1359 rel = heap_open(DatabaseRelationId, RowExclusiveLock);
1360 ScanKeyInit(&scankey,
1361 Anum_pg_database_datname,
1362 BTEqualStrategyNumber, F_NAMEEQ,
1363 NameGetDatum(stmt->dbname));
1364 scan = systable_beginscan(rel, DatabaseNameIndexId, true,
1365 SnapshotNow, 1, &scankey);
1366 tuple = systable_getnext(scan);
1367 if (!HeapTupleIsValid(tuple))
1369 (errcode(ERRCODE_UNDEFINED_DATABASE),
1370 errmsg("database \"%s\" does not exist", stmt->dbname)));
1372 if (!pg_database_ownercheck(HeapTupleGetOid(tuple), GetUserId()))
1373 aclcheck_error(ACLCHECK_NOT_OWNER, ACL_KIND_DATABASE,
1377 * Build an updated tuple, perusing the information just obtained
1379 MemSet(new_record, 0, sizeof(new_record));
1380 MemSet(new_record_nulls, false, sizeof(new_record_nulls));
1381 MemSet(new_record_repl, false, sizeof(new_record_repl));
1385 new_record[Anum_pg_database_datconnlimit - 1] = Int32GetDatum(connlimit);
1386 new_record_repl[Anum_pg_database_datconnlimit - 1] = true;
1389 newtuple = heap_modify_tuple(tuple, RelationGetDescr(rel), new_record,
1390 new_record_nulls, new_record_repl);
1391 simple_heap_update(rel, &tuple->t_self, newtuple);
1393 /* Update indexes */
1394 CatalogUpdateIndexes(rel, newtuple);
1396 systable_endscan(scan);
1398 /* Close pg_database, but keep lock till commit */
1399 heap_close(rel, NoLock);
1404 * ALTER DATABASE name SET ...
1407 AlterDatabaseSet(AlterDatabaseSetStmt *stmt)
1409 Oid datid = get_database_oid(stmt->dbname);
1411 if (!OidIsValid(datid))
1413 (errcode(ERRCODE_UNDEFINED_DATABASE),
1414 errmsg("database \"%s\" does not exist", stmt->dbname)));
1417 * Obtain a lock on the database and make sure it didn't go away in the
1420 shdepLockAndCheckObject(DatabaseRelationId, datid);
1422 if (!pg_database_ownercheck(datid, GetUserId()))
1423 aclcheck_error(ACLCHECK_NOT_OWNER, ACL_KIND_DATABASE,
1426 AlterSetting(datid, InvalidOid, stmt->setstmt);
1428 UnlockSharedObject(DatabaseRelationId, datid, 0, AccessShareLock);
1433 * ALTER DATABASE name OWNER TO newowner
1436 AlterDatabaseOwner(const char *dbname, Oid newOwnerId)
1440 ScanKeyData scankey;
1442 Form_pg_database datForm;
1445 * Get the old tuple. We don't need a lock on the database per se,
1446 * because we're not going to do anything that would mess up incoming
1449 rel = heap_open(DatabaseRelationId, RowExclusiveLock);
1450 ScanKeyInit(&scankey,
1451 Anum_pg_database_datname,
1452 BTEqualStrategyNumber, F_NAMEEQ,
1453 NameGetDatum(dbname));
1454 scan = systable_beginscan(rel, DatabaseNameIndexId, true,
1455 SnapshotNow, 1, &scankey);
1456 tuple = systable_getnext(scan);
1457 if (!HeapTupleIsValid(tuple))
1459 (errcode(ERRCODE_UNDEFINED_DATABASE),
1460 errmsg("database \"%s\" does not exist", dbname)));
1462 datForm = (Form_pg_database) GETSTRUCT(tuple);
1465 * If the new owner is the same as the existing owner, consider the
1466 * command to have succeeded. This is to be consistent with other
1469 if (datForm->datdba != newOwnerId)
1471 Datum repl_val[Natts_pg_database];
1472 bool repl_null[Natts_pg_database];
1473 bool repl_repl[Natts_pg_database];
1479 /* Otherwise, must be owner of the existing object */
1480 if (!pg_database_ownercheck(HeapTupleGetOid(tuple), GetUserId()))
1481 aclcheck_error(ACLCHECK_NOT_OWNER, ACL_KIND_DATABASE,
1484 /* Must be able to become new owner */
1485 check_is_member_of_role(GetUserId(), newOwnerId);
1488 * must have createdb rights
1490 * NOTE: This is different from other alter-owner checks in that the
1491 * current user is checked for createdb privileges instead of the
1492 * destination owner. This is consistent with the CREATE case for
1493 * databases. Because superusers will always have this right, we need
1494 * no special case for them.
1496 if (!have_createdb_privilege())
1498 (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
1499 errmsg("permission denied to change owner of database")));
1501 memset(repl_null, false, sizeof(repl_null));
1502 memset(repl_repl, false, sizeof(repl_repl));
1504 repl_repl[Anum_pg_database_datdba - 1] = true;
1505 repl_val[Anum_pg_database_datdba - 1] = ObjectIdGetDatum(newOwnerId);
1508 * Determine the modified ACL for the new owner. This is only
1509 * necessary when the ACL is non-null.
1511 aclDatum = heap_getattr(tuple,
1512 Anum_pg_database_datacl,
1513 RelationGetDescr(rel),
1517 newAcl = aclnewowner(DatumGetAclP(aclDatum),
1518 datForm->datdba, newOwnerId);
1519 repl_repl[Anum_pg_database_datacl - 1] = true;
1520 repl_val[Anum_pg_database_datacl - 1] = PointerGetDatum(newAcl);
1523 newtuple = heap_modify_tuple(tuple, RelationGetDescr(rel), repl_val, repl_null, repl_repl);
1524 simple_heap_update(rel, &newtuple->t_self, newtuple);
1525 CatalogUpdateIndexes(rel, newtuple);
1527 heap_freetuple(newtuple);
1529 /* Update owner dependency reference */
1530 changeDependencyOnOwner(DatabaseRelationId, HeapTupleGetOid(tuple),
1534 systable_endscan(scan);
1536 /* Close pg_database, but keep lock till commit */
1537 heap_close(rel, NoLock);
1546 * Look up info about the database named "name". If the database exists,
1547 * obtain the specified lock type on it, fill in any of the remaining
1548 * parameters that aren't NULL, and return TRUE. If no such database,
1552 get_db_info(const char *name, LOCKMODE lockmode,
1553 Oid *dbIdP, Oid *ownerIdP,
1554 int *encodingP, bool *dbIsTemplateP, bool *dbAllowConnP,
1555 Oid *dbLastSysOidP, TransactionId *dbFrozenXidP,
1556 Oid *dbTablespace, char **dbCollate, char **dbCtype)
1558 bool result = false;
1563 /* Caller may wish to grab a better lock on pg_database beforehand... */
1564 relation = heap_open(DatabaseRelationId, AccessShareLock);
1567 * Loop covers the rare case where the database is renamed before we can
1568 * lock it. We try again just in case we can find a new one of the same
1573 ScanKeyData scanKey;
1579 * there's no syscache for database-indexed-by-name, so must do it the
1582 ScanKeyInit(&scanKey,
1583 Anum_pg_database_datname,
1584 BTEqualStrategyNumber, F_NAMEEQ,
1585 NameGetDatum(name));
1587 scan = systable_beginscan(relation, DatabaseNameIndexId, true,
1588 SnapshotNow, 1, &scanKey);
1590 tuple = systable_getnext(scan);
1592 if (!HeapTupleIsValid(tuple))
1594 /* definitely no database of that name */
1595 systable_endscan(scan);
1599 dbOid = HeapTupleGetOid(tuple);
1601 systable_endscan(scan);
1604 * Now that we have a database OID, we can try to lock the DB.
1606 if (lockmode != NoLock)
1607 LockSharedObject(DatabaseRelationId, dbOid, 0, lockmode);
1610 * And now, re-fetch the tuple by OID. If it's still there and still
1611 * the same name, we win; else, drop the lock and loop back to try
1614 tuple = SearchSysCache(DATABASEOID,
1615 ObjectIdGetDatum(dbOid),
1617 if (HeapTupleIsValid(tuple))
1619 Form_pg_database dbform = (Form_pg_database) GETSTRUCT(tuple);
1621 if (strcmp(name, NameStr(dbform->datname)) == 0)
1623 /* oid of the database */
1626 /* oid of the owner */
1628 *ownerIdP = dbform->datdba;
1629 /* character encoding */
1631 *encodingP = dbform->encoding;
1632 /* allowed as template? */
1634 *dbIsTemplateP = dbform->datistemplate;
1635 /* allowing connections? */
1637 *dbAllowConnP = dbform->datallowconn;
1638 /* last system OID used in database */
1640 *dbLastSysOidP = dbform->datlastsysoid;
1641 /* limit of frozen XIDs */
1643 *dbFrozenXidP = dbform->datfrozenxid;
1644 /* default tablespace for this database */
1646 *dbTablespace = dbform->dattablespace;
1647 /* default locale settings for this database */
1649 *dbCollate = pstrdup(NameStr(dbform->datcollate));
1651 *dbCtype = pstrdup(NameStr(dbform->datctype));
1652 ReleaseSysCache(tuple);
1656 /* can only get here if it was just renamed */
1657 ReleaseSysCache(tuple);
1660 if (lockmode != NoLock)
1661 UnlockSharedObject(DatabaseRelationId, dbOid, 0, lockmode);
1664 heap_close(relation, AccessShareLock);
1669 /* Check if current user has createdb privileges */
1671 have_createdb_privilege(void)
1673 bool result = false;
1676 /* Superusers can always do everything */
1680 utup = SearchSysCache(AUTHOID,
1681 ObjectIdGetDatum(GetUserId()),
1683 if (HeapTupleIsValid(utup))
1685 result = ((Form_pg_authid) GETSTRUCT(utup))->rolcreatedb;
1686 ReleaseSysCache(utup);
1692 * Remove tablespace directories
1694 * We don't know what tablespaces db_id is using, so iterate through all
1695 * tablespaces removing <tablespace>/db_id
1698 remove_dbtablespaces(Oid db_id)
1704 rel = heap_open(TableSpaceRelationId, AccessShareLock);
1705 scan = heap_beginscan(rel, SnapshotNow, 0, NULL);
1706 while ((tuple = heap_getnext(scan, ForwardScanDirection)) != NULL)
1708 Oid dsttablespace = HeapTupleGetOid(tuple);
1712 /* Don't mess with the global tablespace */
1713 if (dsttablespace == GLOBALTABLESPACE_OID)
1716 dstpath = GetDatabasePath(db_id, dsttablespace);
1718 if (lstat(dstpath, &st) < 0 || !S_ISDIR(st.st_mode))
1720 /* Assume we can ignore it */
1725 if (!rmtree(dstpath, true))
1727 (errmsg("some useless files may be left behind in old database directory \"%s\"",
1730 /* Record the filesystem change in XLOG */
1732 xl_dbase_drop_rec xlrec;
1733 XLogRecData rdata[1];
1735 xlrec.db_id = db_id;
1736 xlrec.tablespace_id = dsttablespace;
1738 rdata[0].data = (char *) &xlrec;
1739 rdata[0].len = sizeof(xl_dbase_drop_rec);
1740 rdata[0].buffer = InvalidBuffer;
1741 rdata[0].next = NULL;
1743 (void) XLogInsert(RM_DBASE_ID, XLOG_DBASE_DROP, rdata);
1750 heap_close(rel, AccessShareLock);
1754 * Check for existing files that conflict with a proposed new DB OID;
1755 * return TRUE if there are any
1757 * If there were a subdirectory in any tablespace matching the proposed new
1758 * OID, we'd get a create failure due to the duplicate name ... and then we'd
1759 * try to remove that already-existing subdirectory during the cleanup in
1760 * remove_dbtablespaces. Nuking existing files seems like a bad idea, so
1761 * instead we make this extra check before settling on the OID of the new
1762 * database. This exactly parallels what GetNewRelFileNode() does for table
1763 * relfilenode values.
1766 check_db_file_conflict(Oid db_id)
1768 bool result = false;
1773 rel = heap_open(TableSpaceRelationId, AccessShareLock);
1774 scan = heap_beginscan(rel, SnapshotNow, 0, NULL);
1775 while ((tuple = heap_getnext(scan, ForwardScanDirection)) != NULL)
1777 Oid dsttablespace = HeapTupleGetOid(tuple);
1781 /* Don't mess with the global tablespace */
1782 if (dsttablespace == GLOBALTABLESPACE_OID)
1785 dstpath = GetDatabasePath(db_id, dsttablespace);
1787 if (lstat(dstpath, &st) == 0)
1789 /* Found a conflicting file (or directory, whatever) */
1799 heap_close(rel, AccessShareLock);
1804 * Issue a suitable errdetail message for a busy database
1807 errdetail_busy_db(int notherbackends, int npreparedxacts)
1810 * We don't worry about singular versus plural here, since the English
1811 * rules for that don't translate very well. But we can at least avoid
1812 * the case of zero items.
1814 if (notherbackends > 0 && npreparedxacts > 0)
1815 errdetail("There are %d other session(s) and %d prepared transaction(s) using the database.",
1816 notherbackends, npreparedxacts);
1817 else if (notherbackends > 0)
1818 errdetail("There are %d other session(s) using the database.",
1821 errdetail("There are %d prepared transaction(s) using the database.",
1823 return 0; /* just to keep ereport macro happy */
1827 * get_database_oid - given a database name, look up the OID
1829 * Returns InvalidOid if database name not found.
1832 get_database_oid(const char *dbname)
1834 Relation pg_database;
1835 ScanKeyData entry[1];
1841 * There's no syscache for pg_database indexed by name, so we must look
1844 pg_database = heap_open(DatabaseRelationId, AccessShareLock);
1845 ScanKeyInit(&entry[0],
1846 Anum_pg_database_datname,
1847 BTEqualStrategyNumber, F_NAMEEQ,
1848 CStringGetDatum(dbname));
1849 scan = systable_beginscan(pg_database, DatabaseNameIndexId, true,
1850 SnapshotNow, 1, entry);
1852 dbtuple = systable_getnext(scan);
1854 /* We assume that there can be at most one matching tuple */
1855 if (HeapTupleIsValid(dbtuple))
1856 oid = HeapTupleGetOid(dbtuple);
1860 systable_endscan(scan);
1861 heap_close(pg_database, AccessShareLock);
1868 * get_database_name - given a database OID, look up the name
1870 * Returns a palloc'd string, or NULL if no such database.
1873 get_database_name(Oid dbid)
1878 dbtuple = SearchSysCache(DATABASEOID,
1879 ObjectIdGetDatum(dbid),
1881 if (HeapTupleIsValid(dbtuple))
1883 result = pstrdup(NameStr(((Form_pg_database) GETSTRUCT(dbtuple))->datname));
1884 ReleaseSysCache(dbtuple);
1893 * DATABASE resource manager's routines
1896 dbase_redo(XLogRecPtr lsn, XLogRecord *record)
1898 uint8 info = record->xl_info & ~XLR_INFO_MASK;
1900 /* Backup blocks are not used in dbase records */
1901 Assert(!(record->xl_info & XLR_BKP_BLOCK_MASK));
1903 if (info == XLOG_DBASE_CREATE)
1905 xl_dbase_create_rec *xlrec = (xl_dbase_create_rec *) XLogRecGetData(record);
1910 src_path = GetDatabasePath(xlrec->src_db_id, xlrec->src_tablespace_id);
1911 dst_path = GetDatabasePath(xlrec->db_id, xlrec->tablespace_id);
1914 * Our theory for replaying a CREATE is to forcibly drop the target
1915 * subdirectory if present, then re-copy the source data. This may be
1916 * more work than needed, but it is simple to implement.
1918 if (stat(dst_path, &st) == 0 && S_ISDIR(st.st_mode))
1920 if (!rmtree(dst_path, true))
1922 (errmsg("some useless files may be left behind in old database directory \"%s\"",
1927 * Force dirty buffers out to disk, to ensure source database is
1928 * up-to-date for the copy.
1930 FlushDatabaseBuffers(xlrec->src_db_id);
1933 * Copy this subdirectory to the new location
1935 * We don't need to copy subdirectories
1937 copydir(src_path, dst_path, false);
1939 else if (info == XLOG_DBASE_DROP)
1941 xl_dbase_drop_rec *xlrec = (xl_dbase_drop_rec *) XLogRecGetData(record);
1944 dst_path = GetDatabasePath(xlrec->db_id, xlrec->tablespace_id);
1948 VirtualTransactionId *database_users;
1951 * Find all users connected to this database and ask them
1952 * politely to immediately kill their sessions before processing
1953 * the drop database record, after the usual grace period.
1954 * We don't wait for commit because drop database is
1955 * non-transactional.
1957 database_users = GetConflictingVirtualXIDs(InvalidTransactionId,
1961 ResolveRecoveryConflictWithVirtualXIDs(database_users,
1963 CONFLICT_MODE_FATAL);
1966 /* Drop pages for this database that are in the shared buffer cache */
1967 DropDatabaseBuffers(xlrec->db_id);
1969 /* Also, clean out any fsync requests that might be pending in md.c */
1970 ForgetDatabaseFsyncRequests(xlrec->db_id);
1972 /* Clean out the xlog relcache too */
1973 XLogDropDatabase(xlrec->db_id);
1975 /* And remove the physical files */
1976 if (!rmtree(dst_path, true))
1978 (errmsg("some useless files may be left behind in old database directory \"%s\"",
1982 elog(PANIC, "dbase_redo: unknown op code %u", info);
1986 dbase_desc(StringInfo buf, uint8 xl_info, char *rec)
1988 uint8 info = xl_info & ~XLR_INFO_MASK;
1990 if (info == XLOG_DBASE_CREATE)
1992 xl_dbase_create_rec *xlrec = (xl_dbase_create_rec *) rec;
1994 appendStringInfo(buf, "create db: copy dir %u/%u to %u/%u",
1995 xlrec->src_db_id, xlrec->src_tablespace_id,
1996 xlrec->db_id, xlrec->tablespace_id);
1998 else if (info == XLOG_DBASE_DROP)
2000 xl_dbase_drop_rec *xlrec = (xl_dbase_drop_rec *) rec;
2002 appendStringInfo(buf, "drop db: dir %u/%u",
2003 xlrec->db_id, xlrec->tablespace_id);
2006 appendStringInfo(buf, "UNKNOWN");