?  search:
summary | log | graphiclog1 | graphiclog2 | commit | commitdiff | tree | refs | edit | fork
blame | history | raw | HEAD
Update copyright for 2022
[pgsql.git] / src / backend / commands / dbcommands.c
blob509d1a3e92f0e382648599662a0c9d75ba8c9c38
1 /*-------------------------------------------------------------------------
2 *
3 * dbcommands.c
4 * Database management commands (create/drop database).
5 *
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.
10 *
11 * Portions Copyright (c) 1996-2022, PostgreSQL Global Development Group
12 * Portions Copyright (c) 1994, Regents of the University of California
13 *
14 *
15 * IDENTIFICATION
16 * src/backend/commands/dbcommands.c
17 *
18 *-------------------------------------------------------------------------
19 */
20 #include "postgres.h"
21
22 #include <fcntl.h>
23 #include <unistd.h>
24 #include <sys/stat.h>
25
26 #include "access/genam.h"
27 #include "access/heapam.h"
28 #include "access/htup_details.h"
29 #include "access/multixact.h"
30 #include "access/tableam.h"
31 #include "access/xact.h"
32 #include "access/xloginsert.h"
33 #include "access/xlogutils.h"
34 #include "catalog/catalog.h"
35 #include "catalog/dependency.h"
36 #include "catalog/indexing.h"
37 #include "catalog/objectaccess.h"
38 #include "catalog/pg_authid.h"
39 #include "catalog/pg_database.h"
40 #include "catalog/pg_db_role_setting.h"
41 #include "catalog/pg_subscription.h"
42 #include "catalog/pg_tablespace.h"
43 #include "commands/comment.h"
44 #include "commands/dbcommands.h"
45 #include "commands/dbcommands_xlog.h"
46 #include "commands/defrem.h"
47 #include "commands/seclabel.h"
48 #include "commands/tablespace.h"
49 #include "mb/pg_wchar.h"
50 #include "miscadmin.h"
51 #include "pgstat.h"
52 #include "postmaster/bgwriter.h"
53 #include "replication/slot.h"
54 #include "storage/copydir.h"
55 #include "storage/fd.h"
56 #include "storage/ipc.h"
57 #include "storage/lmgr.h"
58 #include "storage/md.h"
59 #include "storage/procarray.h"
60 #include "storage/smgr.h"
61 #include "utils/acl.h"
62 #include "utils/builtins.h"
63 #include "utils/fmgroids.h"
64 #include "utils/pg_locale.h"
65 #include "utils/snapmgr.h"
66 #include "utils/syscache.h"
67
68 typedef struct
69 {
70 Oid src_dboid; /* source (template) DB */
71 Oid dest_dboid; /* DB we are trying to create */
72 } createdb_failure_params;
73
74 typedef struct
75 {
76 Oid dest_dboid; /* DB we are trying to move */
77 Oid dest_tsoid; /* tablespace we are trying to move to */
78 } movedb_failure_params;
79
80 /* non-export function prototypes */
81 static void createdb_failure_callback(int code, Datum arg);
82 static void movedb(const char *dbname, const char *tblspcname);
83 static void movedb_failure_callback(int code, Datum arg);
84 static bool get_db_info(const char *name, LOCKMODE lockmode,
85 Oid *dbIdP, Oid *ownerIdP,
86 int *encodingP, bool *dbIsTemplateP, bool *dbAllowConnP,
87 Oid *dbLastSysOidP, TransactionId *dbFrozenXidP,
88 MultiXactId *dbMinMultiP,
89 Oid *dbTablespace, char **dbCollate, char **dbCtype);
90 static bool have_createdb_privilege(void);
91 static void remove_dbtablespaces(Oid db_id);
92 static bool check_db_file_conflict(Oid db_id);
93 static int errdetail_busy_db(int notherbackends, int npreparedxacts);
94
95
96 /*
97 * CREATE DATABASE
98 */
99 Oid
100 createdb(ParseState *pstate, const CreatedbStmt *stmt)
101 {
102 TableScanDesc scan;
103 Relation rel;
104 Oid src_dboid;
105 Oid src_owner;
106 int src_encoding = -1;
107 char *src_collate = NULL;
108 char *src_ctype = NULL;
109 bool src_istemplate;
110 bool src_allowconn;
111 Oid src_lastsysoid = InvalidOid;
112 TransactionId src_frozenxid = InvalidTransactionId;
113 MultiXactId src_minmxid = InvalidMultiXactId;
114 Oid src_deftablespace;
115 volatile Oid dst_deftablespace;
116 Relation pg_database_rel;
117 HeapTuple tuple;
118 Datum new_record[Natts_pg_database];
119 bool new_record_nulls[Natts_pg_database];
120 Oid dboid;
121 Oid datdba;
122 ListCell *option;
123 DefElem *dtablespacename = NULL;
124 DefElem *downer = NULL;
125 DefElem *dtemplate = NULL;
126 DefElem *dencoding = NULL;
127 DefElem *dlocale = NULL;
128 DefElem *dcollate = NULL;
129 DefElem *dctype = NULL;
130 DefElem *distemplate = NULL;
131 DefElem *dallowconnections = NULL;
132 DefElem *dconnlimit = NULL;
133 char *dbname = stmt->dbname;
134 char *dbowner = NULL;
135 const char *dbtemplate = NULL;
136 char *dbcollate = NULL;
137 char *dbctype = NULL;
138 char *canonname;
139 int encoding = -1;
140 bool dbistemplate = false;
141 bool dballowconnections = true;
142 int dbconnlimit = -1;
143 int notherbackends;
144 int npreparedxacts;
145 createdb_failure_params fparms;
146
147 /* Extract options from the statement node tree */
148 foreach(option, stmt->options)
149 {
150 DefElem *defel = (DefElem *) lfirst(option);
151
152 if (strcmp(defel->defname, "tablespace") == 0)
153 {
154 if (dtablespacename)
155 errorConflictingDefElem(defel, pstate);
156 dtablespacename = defel;
157 }
158 else if (strcmp(defel->defname, "owner") == 0)
159 {
160 if (downer)
161 errorConflictingDefElem(defel, pstate);
162 downer = defel;
163 }
164 else if (strcmp(defel->defname, "template") == 0)
165 {
166 if (dtemplate)
167 errorConflictingDefElem(defel, pstate);
168 dtemplate = defel;
169 }
170 else if (strcmp(defel->defname, "encoding") == 0)
171 {
172 if (dencoding)
173 errorConflictingDefElem(defel, pstate);
174 dencoding = defel;
175 }
176 else if (strcmp(defel->defname, "locale") == 0)
177 {
178 if (dlocale)
179 errorConflictingDefElem(defel, pstate);
180 dlocale = defel;
181 }
182 else if (strcmp(defel->defname, "lc_collate") == 0)
183 {
184 if (dcollate)
185 errorConflictingDefElem(defel, pstate);
186 dcollate = defel;
187 }
188 else if (strcmp(defel->defname, "lc_ctype") == 0)
189 {
190 if (dctype)
191 errorConflictingDefElem(defel, pstate);
192 dctype = defel;
193 }
194 else if (strcmp(defel->defname, "is_template") == 0)
195 {
196 if (distemplate)
197 errorConflictingDefElem(defel, pstate);
198 distemplate = defel;
199 }
200 else if (strcmp(defel->defname, "allow_connections") == 0)
201 {
202 if (dallowconnections)
203 errorConflictingDefElem(defel, pstate);
204 dallowconnections = defel;
205 }
206 else if (strcmp(defel->defname, "connection_limit") == 0)
207 {
208 if (dconnlimit)
209 errorConflictingDefElem(defel, pstate);
210 dconnlimit = defel;
211 }
212 else if (strcmp(defel->defname, "location") == 0)
213 {
214 ereport(WARNING,
215 (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
216 errmsg("LOCATION is not supported anymore"),
217 errhint("Consider using tablespaces instead."),
218 parser_errposition(pstate, defel->location)));
219 }
220 else
221 ereport(ERROR,
222 (errcode(ERRCODE_SYNTAX_ERROR),
223 errmsg("option \"%s\" not recognized", defel->defname),
224 parser_errposition(pstate, defel->location)));
225 }
226
227 if (dlocale && (dcollate || dctype))
228 ereport(ERROR,
229 (errcode(ERRCODE_SYNTAX_ERROR),
230 errmsg("conflicting or redundant options"),
231 errdetail("LOCALE cannot be specified together with LC_COLLATE or LC_CTYPE.")));
232
233 if (downer && downer->arg)
234 dbowner = defGetString(downer);
235 if (dtemplate && dtemplate->arg)
236 dbtemplate = defGetString(dtemplate);
237 if (dencoding && dencoding->arg)
238 {
239 const char *encoding_name;
240
241 if (IsA(dencoding->arg, Integer))
242 {
243 encoding = defGetInt32(dencoding);
244 encoding_name = pg_encoding_to_char(encoding);
245 if (strcmp(encoding_name, "") == 0 ||
246 pg_valid_server_encoding(encoding_name) < 0)
247 ereport(ERROR,
248 (errcode(ERRCODE_UNDEFINED_OBJECT),
249 errmsg("%d is not a valid encoding code",
250 encoding),
251 parser_errposition(pstate, dencoding->location)));
252 }
253 else
254 {
255 encoding_name = defGetString(dencoding);
256 encoding = pg_valid_server_encoding(encoding_name);
257 if (encoding < 0)
258 ereport(ERROR,
259 (errcode(ERRCODE_UNDEFINED_OBJECT),
260 errmsg("%s is not a valid encoding name",
261 encoding_name),
262 parser_errposition(pstate, dencoding->location)));
263 }
264 }
265 if (dlocale && dlocale->arg)
266 {
267 dbcollate = defGetString(dlocale);
268 dbctype = defGetString(dlocale);
269 }
270 if (dcollate && dcollate->arg)
271 dbcollate = defGetString(dcollate);
272 if (dctype && dctype->arg)
273 dbctype = defGetString(dctype);
274 if (distemplate && distemplate->arg)
275 dbistemplate = defGetBoolean(distemplate);
276 if (dallowconnections && dallowconnections->arg)
277 dballowconnections = defGetBoolean(dallowconnections);
278 if (dconnlimit && dconnlimit->arg)
279 {
280 dbconnlimit = defGetInt32(dconnlimit);
281 if (dbconnlimit < -1)
282 ereport(ERROR,
283 (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
284 errmsg("invalid connection limit: %d", dbconnlimit)));
285 }
286
287 /* obtain OID of proposed owner */
288 if (dbowner)
289 datdba = get_role_oid(dbowner, false);
290 else
291 datdba = GetUserId();
292
293 /*
294 * To create a database, must have createdb privilege and must be able to
295 * become the target role (this does not imply that the target role itself
296 * must have createdb privilege). The latter provision guards against
297 * "giveaway" attacks. Note that a superuser will always have both of
298 * these privileges a fortiori.
299 */
300 if (!have_createdb_privilege())
301 ereport(ERROR,
302 (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
303 errmsg("permission denied to create database")));
304
305 check_is_member_of_role(GetUserId(), datdba);
306
307 /*
308 * Lookup database (template) to be cloned, and obtain share lock on it.
309 * ShareLock allows two CREATE DATABASEs to work from the same template
310 * concurrently, while ensuring no one is busy dropping it in parallel
311 * (which would be Very Bad since we'd likely get an incomplete copy
312 * without knowing it). This also prevents any new connections from being
313 * made to the source until we finish copying it, so we can be sure it
314 * won't change underneath us.
315 */
316 if (!dbtemplate)
317 dbtemplate = "template1"; /* Default template database name */
318
319 if (!get_db_info(dbtemplate, ShareLock,
320 &src_dboid, &src_owner, &src_encoding,
321 &src_istemplate, &src_allowconn, &src_lastsysoid,
322 &src_frozenxid, &src_minmxid, &src_deftablespace,
323 &src_collate, &src_ctype))
324 ereport(ERROR,
325 (errcode(ERRCODE_UNDEFINED_DATABASE),
326 errmsg("template database \"%s\" does not exist",
327 dbtemplate)));
328
329 /*
330 * Permission check: to copy a DB that's not marked datistemplate, you
331 * must be superuser or the owner thereof.
332 */
333 if (!src_istemplate)
334 {
335 if (!pg_database_ownercheck(src_dboid, GetUserId()))
336 ereport(ERROR,
337 (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
338 errmsg("permission denied to copy database \"%s\"",
339 dbtemplate)));
340 }
341
342 /* If encoding or locales are defaulted, use source's setting */
343 if (encoding < 0)
344 encoding = src_encoding;
345 if (dbcollate == NULL)
346 dbcollate = src_collate;
347 if (dbctype == NULL)
348 dbctype = src_ctype;
349
350 /* Some encodings are client only */
351 if (!PG_VALID_BE_ENCODING(encoding))
352 ereport(ERROR,
353 (errcode(ERRCODE_WRONG_OBJECT_TYPE),
354 errmsg("invalid server encoding %d", encoding)));
355
356 /* Check that the chosen locales are valid, and get canonical spellings */
357 if (!check_locale(LC_COLLATE, dbcollate, &canonname))
358 ereport(ERROR,
359 (errcode(ERRCODE_WRONG_OBJECT_TYPE),
360 errmsg("invalid locale name: \"%s\"", dbcollate)));
361 dbcollate = canonname;
362 if (!check_locale(LC_CTYPE, dbctype, &canonname))
363 ereport(ERROR,
364 (errcode(ERRCODE_WRONG_OBJECT_TYPE),
365 errmsg("invalid locale name: \"%s\"", dbctype)));
366 dbctype = canonname;
367
368 check_encoding_locale_matches(encoding, dbcollate, dbctype);
369
370 /*
371 * Check that the new encoding and locale settings match the source
372 * database. We insist on this because we simply copy the source data ---
373 * any non-ASCII data would be wrongly encoded, and any indexes sorted
374 * according to the source locale would be wrong.
375 *
376 * However, we assume that template0 doesn't contain any non-ASCII data
377 * nor any indexes that depend on collation or ctype, so template0 can be
378 * used as template for creating a database with any encoding or locale.
379 */
380 if (strcmp(dbtemplate, "template0") != 0)
381 {
382 if (encoding != src_encoding)
383 ereport(ERROR,
384 (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
385 errmsg("new encoding (%s) is incompatible with the encoding of the template database (%s)",
386 pg_encoding_to_char(encoding),
387 pg_encoding_to_char(src_encoding)),
388 errhint("Use the same encoding as in the template database, or use template0 as template.")));
389
390 if (strcmp(dbcollate, src_collate) != 0)
391 ereport(ERROR,
392 (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
393 errmsg("new collation (%s) is incompatible with the collation of the template database (%s)",
394 dbcollate, src_collate),
395 errhint("Use the same collation as in the template database, or use template0 as template.")));
396
397 if (strcmp(dbctype, src_ctype) != 0)
398 ereport(ERROR,
399 (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
400 errmsg("new LC_CTYPE (%s) is incompatible with the LC_CTYPE of the template database (%s)",
401 dbctype, src_ctype),
402 errhint("Use the same LC_CTYPE as in the template database, or use template0 as template.")));
403 }
404
405 /* Resolve default tablespace for new database */
406 if (dtablespacename && dtablespacename->arg)
407 {
408 char *tablespacename;
409 AclResult aclresult;
410
411 tablespacename = defGetString(dtablespacename);
412 dst_deftablespace = get_tablespace_oid(tablespacename, false);
413 /* check permissions */
414 aclresult = pg_tablespace_aclcheck(dst_deftablespace, GetUserId(),
415 ACL_CREATE);
416 if (aclresult != ACLCHECK_OK)
417 aclcheck_error(aclresult, OBJECT_TABLESPACE,
418 tablespacename);
419
420 /* pg_global must never be the default tablespace */
421 if (dst_deftablespace == GLOBALTABLESPACE_OID)
422 ereport(ERROR,
423 (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
424 errmsg("pg_global cannot be used as default tablespace")));
425
426 /*
427 * If we are trying to change the default tablespace of the template,
428 * we require that the template not have any files in the new default
429 * tablespace. This is necessary because otherwise the copied
430 * database would contain pg_class rows that refer to its default
431 * tablespace both explicitly (by OID) and implicitly (as zero), which
432 * would cause problems. For example another CREATE DATABASE using
433 * the copied database as template, and trying to change its default
434 * tablespace again, would yield outright incorrect results (it would
435 * improperly move tables to the new default tablespace that should
436 * stay in the same tablespace).
437 */
438 if (dst_deftablespace != src_deftablespace)
439 {
440 char *srcpath;
441 struct stat st;
442
443 srcpath = GetDatabasePath(src_dboid, dst_deftablespace);
444
445 if (stat(srcpath, &st) == 0 &&
446 S_ISDIR(st.st_mode) &&
447 !directory_is_empty(srcpath))
448 ereport(ERROR,
449 (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
450 errmsg("cannot assign new default tablespace \"%s\"",
451 tablespacename),
452 errdetail("There is a conflict because database \"%s\" already has some tables in this tablespace.",
453 dbtemplate)));
454 pfree(srcpath);
455 }
456 }
457 else
458 {
459 /* Use template database's default tablespace */
460 dst_deftablespace = src_deftablespace;
461 /* Note there is no additional permission check in this path */
462 }
463
464 /*
465 * If built with appropriate switch, whine when regression-testing
466 * conventions for database names are violated. But don't complain during
467 * initdb.
468 */
469 #ifdef ENFORCE_REGRESSION_TEST_NAME_RESTRICTIONS
470 if (IsUnderPostmaster && strstr(dbname, "regression") == NULL)
471 elog(WARNING, "databases created by regression test cases should have names including \"regression\"");
472 #endif
473
474 /*
475 * Check for db name conflict. This is just to give a more friendly error
476 * message than "unique index violation". There's a race condition but
477 * we're willing to accept the less friendly message in that case.
478 */
479 if (OidIsValid(get_database_oid(dbname, true)))
480 ereport(ERROR,
481 (errcode(ERRCODE_DUPLICATE_DATABASE),
482 errmsg("database \"%s\" already exists", dbname)));
483
484 /*
485 * The source DB can't have any active backends, except this one
486 * (exception is to allow CREATE DB while connected to template1).
487 * Otherwise we might copy inconsistent data.
488 *
489 * This should be last among the basic error checks, because it involves
490 * potential waiting; we may as well throw an error first if we're gonna
491 * throw one.
492 */
493 if (CountOtherDBBackends(src_dboid, &notherbackends, &npreparedxacts))
494 ereport(ERROR,
495 (errcode(ERRCODE_OBJECT_IN_USE),
496 errmsg("source database \"%s\" is being accessed by other users",
497 dbtemplate),
498 errdetail_busy_db(notherbackends, npreparedxacts)));
499
500 /*
501 * Select an OID for the new database, checking that it doesn't have a
502 * filename conflict with anything already existing in the tablespace
503 * directories.
504 */
505 pg_database_rel = table_open(DatabaseRelationId, RowExclusiveLock);
506
507 do
508 {
509 dboid = GetNewOidWithIndex(pg_database_rel, DatabaseOidIndexId,
510 Anum_pg_database_oid);
511 } while (check_db_file_conflict(dboid));
512
513 /*
514 * Insert a new tuple into pg_database. This establishes our ownership of
515 * the new database name (anyone else trying to insert the same name will
516 * block on the unique index, and fail after we commit).
517 */
518
519 /* Form tuple */
520 MemSet(new_record, 0, sizeof(new_record));
521 MemSet(new_record_nulls, false, sizeof(new_record_nulls));
522
523 new_record[Anum_pg_database_oid - 1] = ObjectIdGetDatum(dboid);
524 new_record[Anum_pg_database_datname - 1] =
525 DirectFunctionCall1(namein, CStringGetDatum(dbname));
526 new_record[Anum_pg_database_datdba - 1] = ObjectIdGetDatum(datdba);
527 new_record[Anum_pg_database_encoding - 1] = Int32GetDatum(encoding);
528 new_record[Anum_pg_database_datcollate - 1] =
529 DirectFunctionCall1(namein, CStringGetDatum(dbcollate));
530 new_record[Anum_pg_database_datctype - 1] =
531 DirectFunctionCall1(namein, CStringGetDatum(dbctype));
532 new_record[Anum_pg_database_datistemplate - 1] = BoolGetDatum(dbistemplate);
533 new_record[Anum_pg_database_datallowconn - 1] = BoolGetDatum(dballowconnections);
534 new_record[Anum_pg_database_datconnlimit - 1] = Int32GetDatum(dbconnlimit);
535 new_record[Anum_pg_database_datlastsysoid - 1] = ObjectIdGetDatum(src_lastsysoid);
536 new_record[Anum_pg_database_datfrozenxid - 1] = TransactionIdGetDatum(src_frozenxid);
537 new_record[Anum_pg_database_datminmxid - 1] = TransactionIdGetDatum(src_minmxid);
538 new_record[Anum_pg_database_dattablespace - 1] = ObjectIdGetDatum(dst_deftablespace);
539
540 /*
541 * We deliberately set datacl to default (NULL), rather than copying it
542 * from the template database. Copying it would be a bad idea when the
543 * owner is not the same as the template's owner.
544 */
545 new_record_nulls[Anum_pg_database_datacl - 1] = true;
546
547 tuple = heap_form_tuple(RelationGetDescr(pg_database_rel),
548 new_record, new_record_nulls);
549
550 CatalogTupleInsert(pg_database_rel, tuple);
551
552 /*
553 * Now generate additional catalog entries associated with the new DB
554 */
555
556 /* Register owner dependency */
557 recordDependencyOnOwner(DatabaseRelationId, dboid, datdba);
558
559 /* Create pg_shdepend entries for objects within database */
560 copyTemplateDependencies(src_dboid, dboid);
561
562 /* Post creation hook for new database */
563 InvokeObjectPostCreateHook(DatabaseRelationId, dboid, 0);
564
565 /*
566 * Force a checkpoint before starting the copy. This will force all dirty
567 * buffers, including those of unlogged tables, out to disk, to ensure
568 * source database is up-to-date on disk for the copy.
569 * FlushDatabaseBuffers() would suffice for that, but we also want to
570 * process any pending unlink requests. Otherwise, if a checkpoint
571 * happened while we're copying files, a file might be deleted just when
572 * we're about to copy it, causing the lstat() call in copydir() to fail
573 * with ENOENT.
574 */
575 RequestCheckpoint(CHECKPOINT_IMMEDIATE | CHECKPOINT_FORCE | CHECKPOINT_WAIT
576 | CHECKPOINT_FLUSH_ALL);
577
578 /*
579 * Once we start copying subdirectories, we need to be able to clean 'em
580 * up if we fail. Use an ENSURE block to make sure this happens. (This
581 * is not a 100% solution, because of the possibility of failure during
582 * transaction commit after we leave this routine, but it should handle
583 * most scenarios.)
584 */
585 fparms.src_dboid = src_dboid;
586 fparms.dest_dboid = dboid;
587 PG_ENSURE_ERROR_CLEANUP(createdb_failure_callback,
588 PointerGetDatum(&fparms));
589 {
590 /*
591 * Iterate through all tablespaces of the template database, and copy
592 * each one to the new database.
593 */
594 rel = table_open(TableSpaceRelationId, AccessShareLock);
595 scan = table_beginscan_catalog(rel, 0, NULL);
596 while ((tuple = heap_getnext(scan, ForwardScanDirection)) != NULL)
597 {
598 Form_pg_tablespace spaceform = (Form_pg_tablespace) GETSTRUCT(tuple);
599 Oid srctablespace = spaceform->oid;
600 Oid dsttablespace;
601 char *srcpath;
602 char *dstpath;
603 struct stat st;
604
605 /* No need to copy global tablespace */
606 if (srctablespace == GLOBALTABLESPACE_OID)
607 continue;
608
609 srcpath = GetDatabasePath(src_dboid, srctablespace);
610
611 if (stat(srcpath, &st) < 0 || !S_ISDIR(st.st_mode) ||
612 directory_is_empty(srcpath))
613 {
614 /* Assume we can ignore it */
615 pfree(srcpath);
616 continue;
617 }
618
619 if (srctablespace == src_deftablespace)
620 dsttablespace = dst_deftablespace;
621 else
622 dsttablespace = srctablespace;
623
624 dstpath = GetDatabasePath(dboid, dsttablespace);
625
626 /*
627 * Copy this subdirectory to the new location
628 *
629 * We don't need to copy subdirectories
630 */
631 copydir(srcpath, dstpath, false);
632
633 /* Record the filesystem change in XLOG */
634 {
635 xl_dbase_create_rec xlrec;
636
637 xlrec.db_id = dboid;
638 xlrec.tablespace_id = dsttablespace;
639 xlrec.src_db_id = src_dboid;
640 xlrec.src_tablespace_id = srctablespace;
641
642 XLogBeginInsert();
643 XLogRegisterData((char *) &xlrec, sizeof(xl_dbase_create_rec));
644
645 (void) XLogInsert(RM_DBASE_ID,
646 XLOG_DBASE_CREATE | XLR_SPECIAL_REL_UPDATE);
647 }
648 }
649 table_endscan(scan);
650 table_close(rel, AccessShareLock);
651
652 /*
653 * We force a checkpoint before committing. This effectively means
654 * that committed XLOG_DBASE_CREATE operations will never need to be
655 * replayed (at least not in ordinary crash recovery; we still have to
656 * make the XLOG entry for the benefit of PITR operations). This
657 * avoids two nasty scenarios:
658 *
659 * #1: When PITR is off, we don't XLOG the contents of newly created
660 * indexes; therefore the drop-and-recreate-whole-directory behavior
661 * of DBASE_CREATE replay would lose such indexes.
662 *
663 * #2: Since we have to recopy the source database during DBASE_CREATE
664 * replay, we run the risk of copying changes in it that were
665 * committed after the original CREATE DATABASE command but before the
666 * system crash that led to the replay. This is at least unexpected
667 * and at worst could lead to inconsistencies, eg duplicate table
668 * names.
669 *
670 * (Both of these were real bugs in releases 8.0 through 8.0.3.)
671 *
672 * In PITR replay, the first of these isn't an issue, and the second
673 * is only a risk if the CREATE DATABASE and subsequent template
674 * database change both occur while a base backup is being taken.
675 * There doesn't seem to be much we can do about that except document
676 * it as a limitation.
677 *
678 * Perhaps if we ever implement CREATE DATABASE in a less cheesy way,
679 * we can avoid this.
680 */
681 RequestCheckpoint(CHECKPOINT_IMMEDIATE | CHECKPOINT_FORCE | CHECKPOINT_WAIT);
682
683 /*
684 * Close pg_database, but keep lock till commit.
685 */
686 table_close(pg_database_rel, NoLock);
687
688 /*
689 * Force synchronous commit, thus minimizing the window between
690 * creation of the database files and committal of the transaction. If
691 * we crash before committing, we'll have a DB that's taking up disk
692 * space but is not in pg_database, which is not good.
693 */
694 ForceSyncCommit();
695 }
696 PG_END_ENSURE_ERROR_CLEANUP(createdb_failure_callback,
697 PointerGetDatum(&fparms));
698
699 return dboid;
700 }
701
702 /*
703 * Check whether chosen encoding matches chosen locale settings. This
704 * restriction is necessary because libc's locale-specific code usually
705 * fails when presented with data in an encoding it's not expecting. We
706 * allow mismatch in four cases:
707 *
708 * 1. locale encoding = SQL_ASCII, which means that the locale is C/POSIX
709 * which works with any encoding.
710 *
711 * 2. locale encoding = -1, which means that we couldn't determine the
712 * locale's encoding and have to trust the user to get it right.
713 *
714 * 3. selected encoding is UTF8 and platform is win32. This is because
715 * UTF8 is a pseudo codepage that is supported in all locales since it's
716 * converted to UTF16 before being used.
717 *
718 * 4. selected encoding is SQL_ASCII, but only if you're a superuser. This
719 * is risky but we have historically allowed it --- notably, the
720 * regression tests require it.
721 *
722 * Note: if you change this policy, fix initdb to match.
723 */
724 void
725 check_encoding_locale_matches(int encoding, const char *collate, const char *ctype)
726 {
727 int ctype_encoding = pg_get_encoding_from_locale(ctype, true);
728 int collate_encoding = pg_get_encoding_from_locale(collate, true);
729
730 if (!(ctype_encoding == encoding ||
731 ctype_encoding == PG_SQL_ASCII ||
732 ctype_encoding == -1 ||
733 #ifdef WIN32
734 encoding == PG_UTF8 ||
735 #endif
736 (encoding == PG_SQL_ASCII && superuser())))
737 ereport(ERROR,
738 (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
739 errmsg("encoding \"%s\" does not match locale \"%s\"",
740 pg_encoding_to_char(encoding),
741 ctype),
742 errdetail("The chosen LC_CTYPE setting requires encoding \"%s\".",
743 pg_encoding_to_char(ctype_encoding))));
744
745 if (!(collate_encoding == encoding ||
746 collate_encoding == PG_SQL_ASCII ||
747 collate_encoding == -1 ||
748 #ifdef WIN32
749 encoding == PG_UTF8 ||
750 #endif
751 (encoding == PG_SQL_ASCII && superuser())))
752 ereport(ERROR,
753 (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
754 errmsg("encoding \"%s\" does not match locale \"%s\"",
755 pg_encoding_to_char(encoding),
756 collate),
757 errdetail("The chosen LC_COLLATE setting requires encoding \"%s\".",
758 pg_encoding_to_char(collate_encoding))));
759 }
760
761 /* Error cleanup callback for createdb */
762 static void
763 createdb_failure_callback(int code, Datum arg)
764 {
765 createdb_failure_params *fparms = (createdb_failure_params *) DatumGetPointer(arg);
766
767 /*
768 * Release lock on source database before doing recursive remove. This is
769 * not essential but it seems desirable to release the lock as soon as
770 * possible.
771 */
772 UnlockSharedObject(DatabaseRelationId, fparms->src_dboid, 0, ShareLock);
773
774 /* Throw away any successfully copied subdirectories */
775 remove_dbtablespaces(fparms->dest_dboid);
776 }
777
778
779 /*
780 * DROP DATABASE
781 */
782 void
783 dropdb(const char *dbname, bool missing_ok, bool force)
784 {
785 Oid db_id;
786 bool db_istemplate;
787 Relation pgdbrel;
788 HeapTuple tup;
789 int notherbackends;
790 int npreparedxacts;
791 int nslots,
792 nslots_active;
793 int nsubscriptions;
794
795 /*
796 * Look up the target database's OID, and get exclusive lock on it. We
797 * need this to ensure that no new backend starts up in the target
798 * database while we are deleting it (see postinit.c), and that no one is
799 * using it as a CREATE DATABASE template or trying to delete it for
800 * themselves.
801 */
802 pgdbrel = table_open(DatabaseRelationId, RowExclusiveLock);
803
804 if (!get_db_info(dbname, AccessExclusiveLock, &db_id, NULL, NULL,
805 &db_istemplate, NULL, NULL, NULL, NULL, NULL, NULL, NULL))
806 {
807 if (!missing_ok)
808 {
809 ereport(ERROR,
810 (errcode(ERRCODE_UNDEFINED_DATABASE),
811 errmsg("database \"%s\" does not exist", dbname)));
812 }
813 else
814 {
815 /* Close pg_database, release the lock, since we changed nothing */
816 table_close(pgdbrel, RowExclusiveLock);
817 ereport(NOTICE,
818 (errmsg("database \"%s\" does not exist, skipping",
819 dbname)));
820 return;
821 }
822 }
823
824 /*
825 * Permission checks
826 */
827 if (!pg_database_ownercheck(db_id, GetUserId()))
828 aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_DATABASE,
829 dbname);
830
831 /* DROP hook for the database being removed */
832 InvokeObjectDropHook(DatabaseRelationId, db_id, 0);
833
834 /*
835 * Disallow dropping a DB that is marked istemplate. This is just to
836 * prevent people from accidentally dropping template0 or template1; they
837 * can do so if they're really determined ...
838 */
839 if (db_istemplate)
840 ereport(ERROR,
841 (errcode(ERRCODE_WRONG_OBJECT_TYPE),
842 errmsg("cannot drop a template database")));
843
844 /* Obviously can't drop my own database */
845 if (db_id == MyDatabaseId)
846 ereport(ERROR,
847 (errcode(ERRCODE_OBJECT_IN_USE),
848 errmsg("cannot drop the currently open database")));
849
850 /*
851 * Check whether there are active logical slots that refer to the
852 * to-be-dropped database. The database lock we are holding prevents the
853 * creation of new slots using the database or existing slots becoming
854 * active.
855 */
856 (void) ReplicationSlotsCountDBSlots(db_id, &nslots, &nslots_active);
857 if (nslots_active)
858 {
859 ereport(ERROR,
860 (errcode(ERRCODE_OBJECT_IN_USE),
861 errmsg("database \"%s\" is used by an active logical replication slot",
862 dbname),
863 errdetail_plural("There is %d active slot.",
864 "There are %d active slots.",
865 nslots_active, nslots_active)));
866 }
867
868 /*
869 * Check if there are subscriptions defined in the target database.
870 *
871 * We can't drop them automatically because they might be holding
872 * resources in other databases/instances.
873 */
874 if ((nsubscriptions = CountDBSubscriptions(db_id)) > 0)
875 ereport(ERROR,
876 (errcode(ERRCODE_OBJECT_IN_USE),
877 errmsg("database \"%s\" is being used by logical replication subscription",
878 dbname),
879 errdetail_plural("There is %d subscription.",
880 "There are %d subscriptions.",
881 nsubscriptions, nsubscriptions)));
882
883
884 /*
885 * Attempt to terminate all existing connections to the target database if
886 * the user has requested to do so.
887 */
888 if (force)
889 TerminateOtherDBBackends(db_id);
890
891 /*
892 * Check for other backends in the target database. (Because we hold the
893 * database lock, no new ones can start after this.)
894 *
895 * As in CREATE DATABASE, check this after other error conditions.
896 */
897 if (CountOtherDBBackends(db_id, &notherbackends, &npreparedxacts))
898 ereport(ERROR,
899 (errcode(ERRCODE_OBJECT_IN_USE),
900 errmsg("database \"%s\" is being accessed by other users",
901 dbname),
902 errdetail_busy_db(notherbackends, npreparedxacts)));
903
904 /*
905 * Remove the database's tuple from pg_database.
906 */
907 tup = SearchSysCache1(DATABASEOID, ObjectIdGetDatum(db_id));
908 if (!HeapTupleIsValid(tup))
909 elog(ERROR, "cache lookup failed for database %u", db_id);
910
911 CatalogTupleDelete(pgdbrel, &tup->t_self);
912
913 ReleaseSysCache(tup);
914
915 /*
916 * Delete any comments or security labels associated with the database.
917 */
918 DeleteSharedComments(db_id, DatabaseRelationId);
919 DeleteSharedSecurityLabel(db_id, DatabaseRelationId);
920
921 /*
922 * Remove settings associated with this database
923 */
924 DropSetting(db_id, InvalidOid);
925
926 /*
927 * Remove shared dependency references for the database.
928 */
929 dropDatabaseDependencies(db_id);
930
931 /*
932 * Drop db-specific replication slots.
933 */
934 ReplicationSlotsDropDBSlots(db_id);
935
936 /*
937 * Drop pages for this database that are in the shared buffer cache. This
938 * is important to ensure that no remaining backend tries to write out a
939 * dirty buffer to the dead database later...
940 */
941 DropDatabaseBuffers(db_id);
942
943 /*
944 * Tell the stats collector to forget it immediately, too.
945 */
946 pgstat_drop_database(db_id);
947
948 /*
949 * Tell checkpointer to forget any pending fsync and unlink requests for
950 * files in the database; else the fsyncs will fail at next checkpoint, or
951 * worse, it will delete files that belong to a newly created database
952 * with the same OID.
953 */
954 ForgetDatabaseSyncRequests(db_id);
955
956 /*
957 * Force a checkpoint to make sure the checkpointer has received the
958 * message sent by ForgetDatabaseSyncRequests. On Windows, this also
959 * ensures that background procs don't hold any open files, which would
960 * cause rmdir() to fail.
961 */
962 RequestCheckpoint(CHECKPOINT_IMMEDIATE | CHECKPOINT_FORCE | CHECKPOINT_WAIT);
963
964 /*
965 * Remove all tablespace subdirs belonging to the database.
966 */
967 remove_dbtablespaces(db_id);
968
969 /*
970 * Close pg_database, but keep lock till commit.
971 */
972 table_close(pgdbrel, NoLock);
973
974 /*
975 * Force synchronous commit, thus minimizing the window between removal of
976 * the database files and committal of the transaction. If we crash before
977 * committing, we'll have a DB that's gone on disk but still there
978 * according to pg_database, which is not good.
979 */
980 ForceSyncCommit();
981 }
982
983
984 /*
985 * Rename database
986 */
987 ObjectAddress
988 RenameDatabase(const char *oldname, const char *newname)
989 {
990 Oid db_id;
991 HeapTuple newtup;
992 Relation rel;
993 int notherbackends;
994 int npreparedxacts;
995 ObjectAddress address;
996
997 /*
998 * Look up the target database's OID, and get exclusive lock on it. We
999 * need this for the same reasons as DROP DATABASE.
1000 */
1001 rel = table_open(DatabaseRelationId, RowExclusiveLock);
1002
1003 if (!get_db_info(oldname, AccessExclusiveLock, &db_id, NULL, NULL,
1004 NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL))
1005 ereport(ERROR,
1006 (errcode(ERRCODE_UNDEFINED_DATABASE),
1007 errmsg("database \"%s\" does not exist", oldname)));
1008
1009 /* must be owner */
1010 if (!pg_database_ownercheck(db_id, GetUserId()))
1011 aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_DATABASE,
1012 oldname);
1013
1014 /* must have createdb rights */
1015 if (!have_createdb_privilege())
1016 ereport(ERROR,
1017 (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
1018 errmsg("permission denied to rename database")));
1019
1020 /*
1021 * If built with appropriate switch, whine when regression-testing
1022 * conventions for database names are violated.
1023 */
1024 #ifdef ENFORCE_REGRESSION_TEST_NAME_RESTRICTIONS
1025 if (strstr(newname, "regression") == NULL)
1026 elog(WARNING, "databases created by regression test cases should have names including \"regression\"");
1027 #endif
1028
1029 /*
1030 * Make sure the new name doesn't exist. See notes for same error in
1031 * CREATE DATABASE.
1032 */
1033 if (OidIsValid(get_database_oid(newname, true)))
1034 ereport(ERROR,
1035 (errcode(ERRCODE_DUPLICATE_DATABASE),
1036 errmsg("database \"%s\" already exists", newname)));
1037
1038 /*
1039 * XXX Client applications probably store the current database somewhere,
1040 * so renaming it could cause confusion. On the other hand, there may not
1041 * be an actual problem besides a little confusion, so think about this
1042 * and decide.
1043 */
1044 if (db_id == MyDatabaseId)
1045 ereport(ERROR,
1046 (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
1047 errmsg("current database cannot be renamed")));
1048
1049 /*
1050 * Make sure the database does not have active sessions. This is the same
1051 * concern as above, but applied to other sessions.
1052 *
1053 * As in CREATE DATABASE, check this after other error conditions.
1054 */
1055 if (CountOtherDBBackends(db_id, &notherbackends, &npreparedxacts))
1056 ereport(ERROR,
1057 (errcode(ERRCODE_OBJECT_IN_USE),
1058 errmsg("database \"%s\" is being accessed by other users",
1059 oldname),
1060 errdetail_busy_db(notherbackends, npreparedxacts)));
1061
1062 /* rename */
1063 newtup = SearchSysCacheCopy1(DATABASEOID, ObjectIdGetDatum(db_id));
1064 if (!HeapTupleIsValid(newtup))
1065 elog(ERROR, "cache lookup failed for database %u", db_id);
1066 namestrcpy(&(((Form_pg_database) GETSTRUCT(newtup))->datname), newname);
1067 CatalogTupleUpdate(rel, &newtup->t_self, newtup);
1068
1069 InvokeObjectPostAlterHook(DatabaseRelationId, db_id, 0);
1070
1071 ObjectAddressSet(address, DatabaseRelationId, db_id);
1072
1073 /*
1074 * Close pg_database, but keep lock till commit.
1075 */
1076 table_close(rel, NoLock);
1077
1078 return address;
1079 }
1080
1081
1082 /*
1083 * ALTER DATABASE SET TABLESPACE
1084 */
1085 static void
1086 movedb(const char *dbname, const char *tblspcname)
1087 {
1088 Oid db_id;
1089 Relation pgdbrel;
1090 int notherbackends;
1091 int npreparedxacts;
1092 HeapTuple oldtuple,
1093 newtuple;
1094 Oid src_tblspcoid,
1095 dst_tblspcoid;
1096 Datum new_record[Natts_pg_database];
1097 bool new_record_nulls[Natts_pg_database];
1098 bool new_record_repl[Natts_pg_database];
1099 ScanKeyData scankey;
1100 SysScanDesc sysscan;
1101 AclResult aclresult;
1102 char *src_dbpath;
1103 char *dst_dbpath;
1104 DIR *dstdir;
1105 struct dirent *xlde;
1106 movedb_failure_params fparms;
1107
1108 /*
1109 * Look up the target database's OID, and get exclusive lock on it. We
1110 * need this to ensure that no new backend starts up in the database while
1111 * we are moving it, and that no one is using it as a CREATE DATABASE
1112 * template or trying to delete it.
1113 */
1114 pgdbrel = table_open(DatabaseRelationId, RowExclusiveLock);
1115
1116 if (!get_db_info(dbname, AccessExclusiveLock, &db_id, NULL, NULL,
1117 NULL, NULL, NULL, NULL, NULL, &src_tblspcoid, NULL, NULL))
1118 ereport(ERROR,
1119 (errcode(ERRCODE_UNDEFINED_DATABASE),
1120 errmsg("database \"%s\" does not exist", dbname)));
1121
1122 /*
1123 * We actually need a session lock, so that the lock will persist across
1124 * the commit/restart below. (We could almost get away with letting the
1125 * lock be released at commit, except that someone could try to move
1126 * relations of the DB back into the old directory while we rmtree() it.)
1127 */
1128 LockSharedObjectForSession(DatabaseRelationId, db_id, 0,
1129 AccessExclusiveLock);
1130
1131 /*
1132 * Permission checks
1133 */
1134 if (!pg_database_ownercheck(db_id, GetUserId()))
1135 aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_DATABASE,
1136 dbname);
1137
1138 /*
1139 * Obviously can't move the tables of my own database
1140 */
1141 if (db_id == MyDatabaseId)
1142 ereport(ERROR,
1143 (errcode(ERRCODE_OBJECT_IN_USE),
1144 errmsg("cannot change the tablespace of the currently open database")));
1145
1146 /*
1147 * Get tablespace's oid
1148 */
1149 dst_tblspcoid = get_tablespace_oid(tblspcname, false);
1150
1151 /*
1152 * Permission checks
1153 */
1154 aclresult = pg_tablespace_aclcheck(dst_tblspcoid, GetUserId(),
1155 ACL_CREATE);
1156 if (aclresult != ACLCHECK_OK)
1157 aclcheck_error(aclresult, OBJECT_TABLESPACE,
1158 tblspcname);
1159
1160 /*
1161 * pg_global must never be the default tablespace
1162 */
1163 if (dst_tblspcoid == GLOBALTABLESPACE_OID)
1164 ereport(ERROR,
1165 (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
1166 errmsg("pg_global cannot be used as default tablespace")));
1167
1168 /*
1169 * No-op if same tablespace
1170 */
1171 if (src_tblspcoid == dst_tblspcoid)
1172 {
1173 table_close(pgdbrel, NoLock);
1174 UnlockSharedObjectForSession(DatabaseRelationId, db_id, 0,
1175 AccessExclusiveLock);
1176 return;
1177 }
1178
1179 /*
1180 * Check for other backends in the target database. (Because we hold the
1181 * database lock, no new ones can start after this.)
1182 *
1183 * As in CREATE DATABASE, check this after other error conditions.
1184 */
1185 if (CountOtherDBBackends(db_id, &notherbackends, &npreparedxacts))
1186 ereport(ERROR,
1187 (errcode(ERRCODE_OBJECT_IN_USE),
1188 errmsg("database \"%s\" is being accessed by other users",
1189 dbname),
1190 errdetail_busy_db(notherbackends, npreparedxacts)));
1191
1192 /*
1193 * Get old and new database paths
1194 */
1195 src_dbpath = GetDatabasePath(db_id, src_tblspcoid);
1196 dst_dbpath = GetDatabasePath(db_id, dst_tblspcoid);
1197
1198 /*
1199 * Force a checkpoint before proceeding. This will force all dirty
1200 * buffers, including those of unlogged tables, out to disk, to ensure
1201 * source database is up-to-date on disk for the copy.
1202 * FlushDatabaseBuffers() would suffice for that, but we also want to
1203 * process any pending unlink requests. Otherwise, the check for existing
1204 * files in the target directory might fail unnecessarily, not to mention
1205 * that the copy might fail due to source files getting deleted under it.
1206 * On Windows, this also ensures that background procs don't hold any open
1207 * files, which would cause rmdir() to fail.
1208 */
1209 RequestCheckpoint(CHECKPOINT_IMMEDIATE | CHECKPOINT_FORCE | CHECKPOINT_WAIT
1210 | CHECKPOINT_FLUSH_ALL);
1211
1212 /*
1213 * Now drop all buffers holding data of the target database; they should
1214 * no longer be dirty so DropDatabaseBuffers is safe.
1215 *
1216 * It might seem that we could just let these buffers age out of shared
1217 * buffers naturally, since they should not get referenced anymore. The
1218 * problem with that is that if the user later moves the database back to
1219 * its original tablespace, any still-surviving buffers would appear to
1220 * contain valid data again --- but they'd be missing any changes made in
1221 * the database while it was in the new tablespace. In any case, freeing
1222 * buffers that should never be used again seems worth the cycles.
1223 *
1224 * Note: it'd be sufficient to get rid of buffers matching db_id and
1225 * src_tblspcoid, but bufmgr.c presently provides no API for that.
1226 */
1227 DropDatabaseBuffers(db_id);
1228
1229 /*
1230 * Check for existence of files in the target directory, i.e., objects of
1231 * this database that are already in the target tablespace. We can't
1232 * allow the move in such a case, because we would need to change those
1233 * relations' pg_class.reltablespace entries to zero, and we don't have
1234 * access to the DB's pg_class to do so.
1235 */
1236 dstdir = AllocateDir(dst_dbpath);
1237 if (dstdir != NULL)
1238 {
1239 while ((xlde = ReadDir(dstdir, dst_dbpath)) != NULL)
1240 {
1241 if (strcmp(xlde->d_name, ".") == 0 ||
1242 strcmp(xlde->d_name, "..") == 0)
1243 continue;
1244
1245 ereport(ERROR,
1246 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1247 errmsg("some relations of database \"%s\" are already in tablespace \"%s\"",
1248 dbname, tblspcname),
1249 errhint("You must move them back to the database's default tablespace before using this command.")));
1250 }
1251
1252 FreeDir(dstdir);
1253
1254 /*
1255 * The directory exists but is empty. We must remove it before using
1256 * the copydir function.
1257 */
1258 if (rmdir(dst_dbpath) != 0)
1259 elog(ERROR, "could not remove directory \"%s\": %m",
1260 dst_dbpath);
1261 }
1262
1263 /*
1264 * Use an ENSURE block to make sure we remove the debris if the copy fails
1265 * (eg, due to out-of-disk-space). This is not a 100% solution, because
1266 * of the possibility of failure during transaction commit, but it should
1267 * handle most scenarios.
1268 */
1269 fparms.dest_dboid = db_id;
1270 fparms.dest_tsoid = dst_tblspcoid;
1271 PG_ENSURE_ERROR_CLEANUP(movedb_failure_callback,
1272 PointerGetDatum(&fparms));
1273 {
1274 /*
1275 * Copy files from the old tablespace to the new one
1276 */
1277 copydir(src_dbpath, dst_dbpath, false);
1278
1279 /*
1280 * Record the filesystem change in XLOG
1281 */
1282 {
1283 xl_dbase_create_rec xlrec;
1284
1285 xlrec.db_id = db_id;
1286 xlrec.tablespace_id = dst_tblspcoid;
1287 xlrec.src_db_id = db_id;
1288 xlrec.src_tablespace_id = src_tblspcoid;
1289
1290 XLogBeginInsert();
1291 XLogRegisterData((char *) &xlrec, sizeof(xl_dbase_create_rec));
1292
1293 (void) XLogInsert(RM_DBASE_ID,
1294 XLOG_DBASE_CREATE | XLR_SPECIAL_REL_UPDATE);
1295 }
1296
1297 /*
1298 * Update the database's pg_database tuple
1299 */
1300 ScanKeyInit(&scankey,
1301 Anum_pg_database_datname,
1302 BTEqualStrategyNumber, F_NAMEEQ,
1303 CStringGetDatum(dbname));
1304 sysscan = systable_beginscan(pgdbrel, DatabaseNameIndexId, true,
1305 NULL, 1, &scankey);
1306 oldtuple = systable_getnext(sysscan);
1307 if (!HeapTupleIsValid(oldtuple)) /* shouldn't happen... */
1308 ereport(ERROR,
1309 (errcode(ERRCODE_UNDEFINED_DATABASE),
1310 errmsg("database \"%s\" does not exist", dbname)));
1311
1312 MemSet(new_record, 0, sizeof(new_record));
1313 MemSet(new_record_nulls, false, sizeof(new_record_nulls));
1314 MemSet(new_record_repl, false, sizeof(new_record_repl));
1315
1316 new_record[Anum_pg_database_dattablespace - 1] = ObjectIdGetDatum(dst_tblspcoid);
1317 new_record_repl[Anum_pg_database_dattablespace - 1] = true;
1318
1319 newtuple = heap_modify_tuple(oldtuple, RelationGetDescr(pgdbrel),
1320 new_record,
1321 new_record_nulls, new_record_repl);
1322 CatalogTupleUpdate(pgdbrel, &oldtuple->t_self, newtuple);
1323
1324 InvokeObjectPostAlterHook(DatabaseRelationId, db_id, 0);
1325
1326 systable_endscan(sysscan);
1327
1328 /*
1329 * Force another checkpoint here. As in CREATE DATABASE, this is to
1330 * ensure that we don't have to replay a committed XLOG_DBASE_CREATE
1331 * operation, which would cause us to lose any unlogged operations
1332 * done in the new DB tablespace before the next checkpoint.
1333 */
1334 RequestCheckpoint(CHECKPOINT_IMMEDIATE | CHECKPOINT_FORCE | CHECKPOINT_WAIT);
1335
1336 /*
1337 * Force synchronous commit, thus minimizing the window between
1338 * copying the database files and committal of the transaction. If we
1339 * crash before committing, we'll leave an orphaned set of files on
1340 * disk, which is not fatal but not good either.
1341 */
1342 ForceSyncCommit();
1343
1344 /*
1345 * Close pg_database, but keep lock till commit.
1346 */
1347 table_close(pgdbrel, NoLock);
1348 }
1349 PG_END_ENSURE_ERROR_CLEANUP(movedb_failure_callback,
1350 PointerGetDatum(&fparms));
1351
1352 /*
1353 * Commit the transaction so that the pg_database update is committed. If
1354 * we crash while removing files, the database won't be corrupt, we'll
1355 * just leave some orphaned files in the old directory.
1356 *
1357 * (This is OK because we know we aren't inside a transaction block.)
1358 *
1359 * XXX would it be safe/better to do this inside the ensure block? Not
1360 * convinced it's a good idea; consider elog just after the transaction
1361 * really commits.
1362 */
1363 PopActiveSnapshot();
1364 CommitTransactionCommand();
1365
1366 /* Start new transaction for the remaining work; don't need a snapshot */
1367 StartTransactionCommand();
1368
1369 /*
1370 * Remove files from the old tablespace
1371 */
1372 if (!rmtree(src_dbpath, true))
1373 ereport(WARNING,
1374 (errmsg("some useless files may be left behind in old database directory \"%s\"",
1375 src_dbpath)));
1376
1377 /*
1378 * Record the filesystem change in XLOG
1379 */
1380 {
1381 xl_dbase_drop_rec xlrec;
1382
1383 xlrec.db_id = db_id;
1384 xlrec.ntablespaces = 1;
1385
1386 XLogBeginInsert();
1387 XLogRegisterData((char *) &xlrec, sizeof(xl_dbase_drop_rec));
1388 XLogRegisterData((char *) &src_tblspcoid, sizeof(Oid));
1389
1390 (void) XLogInsert(RM_DBASE_ID,
1391 XLOG_DBASE_DROP | XLR_SPECIAL_REL_UPDATE);
1392 }
1393
1394 /* Now it's safe to release the database lock */
1395 UnlockSharedObjectForSession(DatabaseRelationId, db_id, 0,
1396 AccessExclusiveLock);
1397 }
1398
1399 /* Error cleanup callback for movedb */
1400 static void
1401 movedb_failure_callback(int code, Datum arg)
1402 {
1403 movedb_failure_params *fparms = (movedb_failure_params *) DatumGetPointer(arg);
1404 char *dstpath;
1405
1406 /* Get rid of anything we managed to copy to the target directory */
1407 dstpath = GetDatabasePath(fparms->dest_dboid, fparms->dest_tsoid);
1408
1409 (void) rmtree(dstpath, true);
1410 }
1411
1412 /*
1413 * Process options and call dropdb function.
1414 */
1415 void
1416 DropDatabase(ParseState *pstate, DropdbStmt *stmt)
1417 {
1418 bool force = false;
1419 ListCell *lc;
1420
1421 foreach(lc, stmt->options)
1422 {
1423 DefElem *opt = (DefElem *) lfirst(lc);
1424
1425 if (strcmp(opt->defname, "force") == 0)
1426 force = true;
1427 else
1428 ereport(ERROR,
1429 (errcode(ERRCODE_SYNTAX_ERROR),
1430 errmsg("unrecognized DROP DATABASE option \"%s\"", opt->defname),
1431 parser_errposition(pstate, opt->location)));
1432 }
1433
1434 dropdb(stmt->dbname, stmt->missing_ok, force);
1435 }
1436
1437 /*
1438 * ALTER DATABASE name ...
1439 */
1440 Oid
1441 AlterDatabase(ParseState *pstate, AlterDatabaseStmt *stmt, bool isTopLevel)
1442 {
1443 Relation rel;
1444 Oid dboid;
1445 HeapTuple tuple,
1446 newtuple;
1447 Form_pg_database datform;
1448 ScanKeyData scankey;
1449 SysScanDesc scan;
1450 ListCell *option;
1451 bool dbistemplate = false;
1452 bool dballowconnections = true;
1453 int dbconnlimit = -1;
1454 DefElem *distemplate = NULL;
1455 DefElem *dallowconnections = NULL;
1456 DefElem *dconnlimit = NULL;
1457 DefElem *dtablespace = NULL;
1458 Datum new_record[Natts_pg_database];
1459 bool new_record_nulls[Natts_pg_database];
1460 bool new_record_repl[Natts_pg_database];
1461
1462 /* Extract options from the statement node tree */
1463 foreach(option, stmt->options)
1464 {
1465 DefElem *defel = (DefElem *) lfirst(option);
1466
1467 if (strcmp(defel->defname, "is_template") == 0)
1468 {
1469 if (distemplate)
1470 errorConflictingDefElem(defel, pstate);
1471 distemplate = defel;
1472 }
1473 else if (strcmp(defel->defname, "allow_connections") == 0)
1474 {
1475 if (dallowconnections)
1476 errorConflictingDefElem(defel, pstate);
1477 dallowconnections = defel;
1478 }
1479 else if (strcmp(defel->defname, "connection_limit") == 0)
1480 {
1481 if (dconnlimit)
1482 errorConflictingDefElem(defel, pstate);
1483 dconnlimit = defel;
1484 }
1485 else if (strcmp(defel->defname, "tablespace") == 0)
1486 {
1487 if (dtablespace)
1488 errorConflictingDefElem(defel, pstate);
1489 dtablespace = defel;
1490 }
1491 else
1492 ereport(ERROR,
1493 (errcode(ERRCODE_SYNTAX_ERROR),
1494 errmsg("option \"%s\" not recognized", defel->defname),
1495 parser_errposition(pstate, defel->location)));
1496 }
1497
1498 if (dtablespace)
1499 {
1500 /*
1501 * While the SET TABLESPACE syntax doesn't allow any other options,
1502 * somebody could write "WITH TABLESPACE ...". Forbid any other
1503 * options from being specified in that case.
1504 */
1505 if (list_length(stmt->options) != 1)
1506 ereport(ERROR,
1507 (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
1508 errmsg("option \"%s\" cannot be specified with other options",
1509 dtablespace->defname),
1510 parser_errposition(pstate, dtablespace->location)));
1511 /* this case isn't allowed within a transaction block */
1512 PreventInTransactionBlock(isTopLevel, "ALTER DATABASE SET TABLESPACE");
1513 movedb(stmt->dbname, defGetString(dtablespace));
1514 return InvalidOid;
1515 }
1516
1517 if (distemplate && distemplate->arg)
1518 dbistemplate = defGetBoolean(distemplate);
1519 if (dallowconnections && dallowconnections->arg)
1520 dballowconnections = defGetBoolean(dallowconnections);
1521 if (dconnlimit && dconnlimit->arg)
1522 {
1523 dbconnlimit = defGetInt32(dconnlimit);
1524 if (dbconnlimit < -1)
1525 ereport(ERROR,
1526 (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
1527 errmsg("invalid connection limit: %d", dbconnlimit)));
1528 }
1529
1530 /*
1531 * Get the old tuple. We don't need a lock on the database per se,
1532 * because we're not going to do anything that would mess up incoming
1533 * connections.
1534 */
1535 rel = table_open(DatabaseRelationId, RowExclusiveLock);
1536 ScanKeyInit(&scankey,
1537 Anum_pg_database_datname,
1538 BTEqualStrategyNumber, F_NAMEEQ,
1539 CStringGetDatum(stmt->dbname));
1540 scan = systable_beginscan(rel, DatabaseNameIndexId, true,
1541 NULL, 1, &scankey);
1542 tuple = systable_getnext(scan);
1543 if (!HeapTupleIsValid(tuple))
1544 ereport(ERROR,
1545 (errcode(ERRCODE_UNDEFINED_DATABASE),
1546 errmsg("database \"%s\" does not exist", stmt->dbname)));
1547
1548 datform = (Form_pg_database) GETSTRUCT(tuple);
1549 dboid = datform->oid;
1550
1551 if (!pg_database_ownercheck(dboid, GetUserId()))
1552 aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_DATABASE,
1553 stmt->dbname);
1554
1555 /*
1556 * In order to avoid getting locked out and having to go through
1557 * standalone mode, we refuse to disallow connections to the database
1558 * we're currently connected to. Lockout can still happen with concurrent
1559 * sessions but the likeliness of that is not high enough to worry about.
1560 */
1561 if (!dballowconnections && dboid == MyDatabaseId)
1562 ereport(ERROR,
1563 (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
1564 errmsg("cannot disallow connections for current database")));
1565
1566 /*
1567 * Build an updated tuple, perusing the information just obtained
1568 */
1569 MemSet(new_record, 0, sizeof(new_record));
1570 MemSet(new_record_nulls, false, sizeof(new_record_nulls));
1571 MemSet(new_record_repl, false, sizeof(new_record_repl));
1572
1573 if (distemplate)
1574 {
1575 new_record[Anum_pg_database_datistemplate - 1] = BoolGetDatum(dbistemplate);
1576 new_record_repl[Anum_pg_database_datistemplate - 1] = true;
1577 }
1578 if (dallowconnections)
1579 {
1580 new_record[Anum_pg_database_datallowconn - 1] = BoolGetDatum(dballowconnections);
1581 new_record_repl[Anum_pg_database_datallowconn - 1] = true;
1582 }
1583 if (dconnlimit)
1584 {
1585 new_record[Anum_pg_database_datconnlimit - 1] = Int32GetDatum(dbconnlimit);
1586 new_record_repl[Anum_pg_database_datconnlimit - 1] = true;
1587 }
1588
1589 newtuple = heap_modify_tuple(tuple, RelationGetDescr(rel), new_record,
1590 new_record_nulls, new_record_repl);
1591 CatalogTupleUpdate(rel, &tuple->t_self, newtuple);
1592
1593 InvokeObjectPostAlterHook(DatabaseRelationId, dboid, 0);
1594
1595 systable_endscan(scan);
1596
1597 /* Close pg_database, but keep lock till commit */
1598 table_close(rel, NoLock);
1599
1600 return dboid;
1601 }
1602
1603
1604 /*
1605 * ALTER DATABASE name SET ...
1606 */
1607 Oid
1608 AlterDatabaseSet(AlterDatabaseSetStmt *stmt)
1609 {
1610 Oid datid = get_database_oid(stmt->dbname, false);
1611
1612 /*
1613 * Obtain a lock on the database and make sure it didn't go away in the
1614 * meantime.
1615 */
1616 shdepLockAndCheckObject(DatabaseRelationId, datid);
1617
1618 if (!pg_database_ownercheck(datid, GetUserId()))
1619 aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_DATABASE,
1620 stmt->dbname);
1621
1622 AlterSetting(datid, InvalidOid, stmt->setstmt);
1623
1624 UnlockSharedObject(DatabaseRelationId, datid, 0, AccessShareLock);
1625
1626 return datid;
1627 }
1628
1629
1630 /*
1631 * ALTER DATABASE name OWNER TO newowner
1632 */
1633 ObjectAddress
1634 AlterDatabaseOwner(const char *dbname, Oid newOwnerId)
1635 {
1636 Oid db_id;
1637 HeapTuple tuple;
1638 Relation rel;
1639 ScanKeyData scankey;
1640 SysScanDesc scan;
1641 Form_pg_database datForm;
1642 ObjectAddress address;
1643
1644 /*
1645 * Get the old tuple. We don't need a lock on the database per se,
1646 * because we're not going to do anything that would mess up incoming
1647 * connections.
1648 */
1649 rel = table_open(DatabaseRelationId, RowExclusiveLock);
1650 ScanKeyInit(&scankey,
1651 Anum_pg_database_datname,
1652 BTEqualStrategyNumber, F_NAMEEQ,
1653 CStringGetDatum(dbname));
1654 scan = systable_beginscan(rel, DatabaseNameIndexId, true,
1655 NULL, 1, &scankey);
1656 tuple = systable_getnext(scan);
1657 if (!HeapTupleIsValid(tuple))
1658 ereport(ERROR,
1659 (errcode(ERRCODE_UNDEFINED_DATABASE),
1660 errmsg("database \"%s\" does not exist", dbname)));
1661
1662 datForm = (Form_pg_database) GETSTRUCT(tuple);
1663 db_id = datForm->oid;
1664
1665 /*
1666 * If the new owner is the same as the existing owner, consider the
1667 * command to have succeeded. This is to be consistent with other
1668 * objects.
1669 */
1670 if (datForm->datdba != newOwnerId)
1671 {
1672 Datum repl_val[Natts_pg_database];
1673 bool repl_null[Natts_pg_database];
1674 bool repl_repl[Natts_pg_database];
1675 Acl *newAcl;
1676 Datum aclDatum;
1677 bool isNull;
1678 HeapTuple newtuple;
1679
1680 /* Otherwise, must be owner of the existing object */
1681 if (!pg_database_ownercheck(db_id, GetUserId()))
1682 aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_DATABASE,
1683 dbname);
1684
1685 /* Must be able to become new owner */
1686 check_is_member_of_role(GetUserId(), newOwnerId);
1687
1688 /*
1689 * must have createdb rights
1690 *
1691 * NOTE: This is different from other alter-owner checks in that the
1692 * current user is checked for createdb privileges instead of the
1693 * destination owner. This is consistent with the CREATE case for
1694 * databases. Because superusers will always have this right, we need
1695 * no special case for them.
1696 */
1697 if (!have_createdb_privilege())
1698 ereport(ERROR,
1699 (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
1700 errmsg("permission denied to change owner of database")));
1701
1702 memset(repl_null, false, sizeof(repl_null));
1703 memset(repl_repl, false, sizeof(repl_repl));
1704
1705 repl_repl[Anum_pg_database_datdba - 1] = true;
1706 repl_val[Anum_pg_database_datdba - 1] = ObjectIdGetDatum(newOwnerId);
1707
1708 /*
1709 * Determine the modified ACL for the new owner. This is only
1710 * necessary when the ACL is non-null.
1711 */
1712 aclDatum = heap_getattr(tuple,
1713 Anum_pg_database_datacl,
1714 RelationGetDescr(rel),
1715 &isNull);
1716 if (!isNull)
1717 {
1718 newAcl = aclnewowner(DatumGetAclP(aclDatum),
1719 datForm->datdba, newOwnerId);
1720 repl_repl[Anum_pg_database_datacl - 1] = true;
1721 repl_val[Anum_pg_database_datacl - 1] = PointerGetDatum(newAcl);
1722 }
1723
1724 newtuple = heap_modify_tuple(tuple, RelationGetDescr(rel), repl_val, repl_null, repl_repl);
1725 CatalogTupleUpdate(rel, &newtuple->t_self, newtuple);
1726
1727 heap_freetuple(newtuple);
1728
1729 /* Update owner dependency reference */
1730 changeDependencyOnOwner(DatabaseRelationId, db_id, newOwnerId);
1731 }
1732
1733 InvokeObjectPostAlterHook(DatabaseRelationId, db_id, 0);
1734
1735 ObjectAddressSet(address, DatabaseRelationId, db_id);
1736
1737 systable_endscan(scan);
1738
1739 /* Close pg_database, but keep lock till commit */
1740 table_close(rel, NoLock);
1741
1742 return address;
1743 }
1744
1745
1746 /*
1747 * Helper functions
1748 */
1749
1750 /*
1751 * Look up info about the database named "name". If the database exists,
1752 * obtain the specified lock type on it, fill in any of the remaining
1753 * parameters that aren't NULL, and return true. If no such database,
1754 * return false.
1755 */
1756 static bool
1757 get_db_info(const char *name, LOCKMODE lockmode,
1758 Oid *dbIdP, Oid *ownerIdP,
1759 int *encodingP, bool *dbIsTemplateP, bool *dbAllowConnP,
1760 Oid *dbLastSysOidP, TransactionId *dbFrozenXidP,
1761 MultiXactId *dbMinMultiP,
1762 Oid *dbTablespace, char **dbCollate, char **dbCtype)
1763 {
1764 bool result = false;
1765 Relation relation;
1766
1767 AssertArg(name);
1768
1769 /* Caller may wish to grab a better lock on pg_database beforehand... */
1770 relation = table_open(DatabaseRelationId, AccessShareLock);
1771
1772 /*
1773 * Loop covers the rare case where the database is renamed before we can
1774 * lock it. We try again just in case we can find a new one of the same
1775 * name.
1776 */
1777 for (;;)
1778 {
1779 ScanKeyData scanKey;
1780 SysScanDesc scan;
1781 HeapTuple tuple;
1782 Oid dbOid;
1783
1784 /*
1785 * there's no syscache for database-indexed-by-name, so must do it the
1786 * hard way
1787 */
1788 ScanKeyInit(&scanKey,
1789 Anum_pg_database_datname,
1790 BTEqualStrategyNumber, F_NAMEEQ,
1791 CStringGetDatum(name));
1792
1793 scan = systable_beginscan(relation, DatabaseNameIndexId, true,
1794 NULL, 1, &scanKey);
1795
1796 tuple = systable_getnext(scan);
1797
1798 if (!HeapTupleIsValid(tuple))
1799 {
1800 /* definitely no database of that name */
1801 systable_endscan(scan);
1802 break;
1803 }
1804
1805 dbOid = ((Form_pg_database) GETSTRUCT(tuple))->oid;
1806
1807 systable_endscan(scan);
1808
1809 /*
1810 * Now that we have a database OID, we can try to lock the DB.
1811 */
1812 if (lockmode != NoLock)
1813 LockSharedObject(DatabaseRelationId, dbOid, 0, lockmode);
1814
1815 /*
1816 * And now, re-fetch the tuple by OID. If it's still there and still
1817 * the same name, we win; else, drop the lock and loop back to try
1818 * again.
1819 */
1820 tuple = SearchSysCache1(DATABASEOID, ObjectIdGetDatum(dbOid));
1821 if (HeapTupleIsValid(tuple))
1822 {
1823 Form_pg_database dbform = (Form_pg_database) GETSTRUCT(tuple);
1824
1825 if (strcmp(name, NameStr(dbform->datname)) == 0)
1826 {
1827 /* oid of the database */
1828 if (dbIdP)
1829 *dbIdP = dbOid;
1830 /* oid of the owner */
1831 if (ownerIdP)
1832 *ownerIdP = dbform->datdba;
1833 /* character encoding */
1834 if (encodingP)
1835 *encodingP = dbform->encoding;
1836 /* allowed as template? */
1837 if (dbIsTemplateP)
1838 *dbIsTemplateP = dbform->datistemplate;
1839 /* allowing connections? */
1840 if (dbAllowConnP)
1841 *dbAllowConnP = dbform->datallowconn;
1842 /* last system OID used in database */
1843 if (dbLastSysOidP)
1844 *dbLastSysOidP = dbform->datlastsysoid;
1845 /* limit of frozen XIDs */
1846 if (dbFrozenXidP)
1847 *dbFrozenXidP = dbform->datfrozenxid;
1848 /* minimum MultiXactId */
1849 if (dbMinMultiP)
1850 *dbMinMultiP = dbform->datminmxid;
1851 /* default tablespace for this database */
1852 if (dbTablespace)
1853 *dbTablespace = dbform->dattablespace;
1854 /* default locale settings for this database */
1855 if (dbCollate)
1856 *dbCollate = pstrdup(NameStr(dbform->datcollate));
1857 if (dbCtype)
1858 *dbCtype = pstrdup(NameStr(dbform->datctype));
1859 ReleaseSysCache(tuple);
1860 result = true;
1861 break;
1862 }
1863 /* can only get here if it was just renamed */
1864 ReleaseSysCache(tuple);
1865 }
1866
1867 if (lockmode != NoLock)
1868 UnlockSharedObject(DatabaseRelationId, dbOid, 0, lockmode);
1869 }
1870
1871 table_close(relation, AccessShareLock);
1872
1873 return result;
1874 }
1875
1876 /* Check if current user has createdb privileges */
1877 static bool
1878 have_createdb_privilege(void)
1879 {
1880 bool result = false;
1881 HeapTuple utup;
1882
1883 /* Superusers can always do everything */
1884 if (superuser())
1885 return true;
1886
1887 utup = SearchSysCache1(AUTHOID, ObjectIdGetDatum(GetUserId()));
1888 if (HeapTupleIsValid(utup))
1889 {
1890 result = ((Form_pg_authid) GETSTRUCT(utup))->rolcreatedb;
1891 ReleaseSysCache(utup);
1892 }
1893 return result;
1894 }
1895
1896 /*
1897 * Remove tablespace directories
1898 *
1899 * We don't know what tablespaces db_id is using, so iterate through all
1900 * tablespaces removing <tablespace>/db_id
1901 */
1902 static void
1903 remove_dbtablespaces(Oid db_id)
1904 {
1905 Relation rel;
1906 TableScanDesc scan;
1907 HeapTuple tuple;
1908 List *ltblspc = NIL;
1909 ListCell *cell;
1910 int ntblspc;
1911 int i;
1912 Oid *tablespace_ids;
1913
1914 rel = table_open(TableSpaceRelationId, AccessShareLock);
1915 scan = table_beginscan_catalog(rel, 0, NULL);
1916 while ((tuple = heap_getnext(scan, ForwardScanDirection)) != NULL)
1917 {
1918 Form_pg_tablespace spcform = (Form_pg_tablespace) GETSTRUCT(tuple);
1919 Oid dsttablespace = spcform->oid;
1920 char *dstpath;
1921 struct stat st;
1922
1923 /* Don't mess with the global tablespace */
1924 if (dsttablespace == GLOBALTABLESPACE_OID)
1925 continue;
1926
1927 dstpath = GetDatabasePath(db_id, dsttablespace);
1928
1929 if (lstat(dstpath, &st) < 0 || !S_ISDIR(st.st_mode))
1930 {
1931 /* Assume we can ignore it */
1932 pfree(dstpath);
1933 continue;
1934 }
1935
1936 if (!rmtree(dstpath, true))
1937 ereport(WARNING,
1938 (errmsg("some useless files may be left behind in old database directory \"%s\"",
1939 dstpath)));
1940
1941 ltblspc = lappend_oid(ltblspc, dsttablespace);
1942 pfree(dstpath);
1943 }
1944
1945 ntblspc = list_length(ltblspc);
1946 if (ntblspc == 0)
1947 {
1948 table_endscan(scan);
1949 table_close(rel, AccessShareLock);
1950 return;
1951 }
1952
1953 tablespace_ids = (Oid *) palloc(ntblspc * sizeof(Oid));
1954 i = 0;
1955 foreach(cell, ltblspc)
1956 tablespace_ids[i++] = lfirst_oid(cell);
1957
1958 /* Record the filesystem change in XLOG */
1959 {
1960 xl_dbase_drop_rec xlrec;
1961
1962 xlrec.db_id = db_id;
1963 xlrec.ntablespaces = ntblspc;
1964
1965 XLogBeginInsert();
1966 XLogRegisterData((char *) &xlrec, MinSizeOfDbaseDropRec);
1967 XLogRegisterData((char *) tablespace_ids, ntblspc * sizeof(Oid));
1968
1969 (void) XLogInsert(RM_DBASE_ID,
1970 XLOG_DBASE_DROP | XLR_SPECIAL_REL_UPDATE);
1971 }
1972
1973 list_free(ltblspc);
1974 pfree(tablespace_ids);
1975
1976 table_endscan(scan);
1977 table_close(rel, AccessShareLock);
1978 }
1979
1980 /*
1981 * Check for existing files that conflict with a proposed new DB OID;
1982 * return true if there are any
1983 *
1984 * If there were a subdirectory in any tablespace matching the proposed new
1985 * OID, we'd get a create failure due to the duplicate name ... and then we'd
1986 * try to remove that already-existing subdirectory during the cleanup in
1987 * remove_dbtablespaces. Nuking existing files seems like a bad idea, so
1988 * instead we make this extra check before settling on the OID of the new
1989 * database. This exactly parallels what GetNewRelFileNode() does for table
1990 * relfilenode values.
1991 */
1992 static bool
1993 check_db_file_conflict(Oid db_id)
1994 {
1995 bool result = false;
1996 Relation rel;
1997 TableScanDesc scan;
1998 HeapTuple tuple;
1999
2000 rel = table_open(TableSpaceRelationId, AccessShareLock);
2001 scan = table_beginscan_catalog(rel, 0, NULL);
2002 while ((tuple = heap_getnext(scan, ForwardScanDirection)) != NULL)
2003 {
2004 Form_pg_tablespace spcform = (Form_pg_tablespace) GETSTRUCT(tuple);
2005 Oid dsttablespace = spcform->oid;
2006 char *dstpath;
2007 struct stat st;
2008
2009 /* Don't mess with the global tablespace */
2010 if (dsttablespace == GLOBALTABLESPACE_OID)
2011 continue;
2012
2013 dstpath = GetDatabasePath(db_id, dsttablespace);
2014
2015 if (lstat(dstpath, &st) == 0)
2016 {
2017 /* Found a conflicting file (or directory, whatever) */
2018 pfree(dstpath);
2019 result = true;
2020 break;
2021 }
2022
2023 pfree(dstpath);
2024 }
2025
2026 table_endscan(scan);
2027 table_close(rel, AccessShareLock);
2028
2029 return result;
2030 }
2031
2032 /*
2033 * Issue a suitable errdetail message for a busy database
2034 */
2035 static int
2036 errdetail_busy_db(int notherbackends, int npreparedxacts)
2037 {
2038 if (notherbackends > 0 && npreparedxacts > 0)
2039
2040 /*
2041 * We don't deal with singular versus plural here, since gettext
2042 * doesn't support multiple plurals in one string.
2043 */
2044 errdetail("There are %d other session(s) and %d prepared transaction(s) using the database.",
2045 notherbackends, npreparedxacts);
2046 else if (notherbackends > 0)
2047 errdetail_plural("There is %d other session using the database.",
2048 "There are %d other sessions using the database.",
2049 notherbackends,
2050 notherbackends);
2051 else
2052 errdetail_plural("There is %d prepared transaction using the database.",
2053 "There are %d prepared transactions using the database.",
2054 npreparedxacts,
2055 npreparedxacts);
2056 return 0; /* just to keep ereport macro happy */
2057 }
2058
2059 /*
2060 * get_database_oid - given a database name, look up the OID
2061 *
2062 * If missing_ok is false, throw an error if database name not found. If
2063 * true, just return InvalidOid.
2064 */
2065 Oid
2066 get_database_oid(const char *dbname, bool missing_ok)
2067 {
2068 Relation pg_database;
2069 ScanKeyData entry[1];
2070 SysScanDesc scan;
2071 HeapTuple dbtuple;
2072 Oid oid;
2073
2074 /*
2075 * There's no syscache for pg_database indexed by name, so we must look
2076 * the hard way.
2077 */
2078 pg_database = table_open(DatabaseRelationId, AccessShareLock);
2079 ScanKeyInit(&entry[0],
2080 Anum_pg_database_datname,
2081 BTEqualStrategyNumber, F_NAMEEQ,
2082 CStringGetDatum(dbname));
2083 scan = systable_beginscan(pg_database, DatabaseNameIndexId, true,
2084 NULL, 1, entry);
2085
2086 dbtuple = systable_getnext(scan);
2087
2088 /* We assume that there can be at most one matching tuple */
2089 if (HeapTupleIsValid(dbtuple))
2090 oid = ((Form_pg_database) GETSTRUCT(dbtuple))->oid;
2091 else
2092 oid = InvalidOid;
2093
2094 systable_endscan(scan);
2095 table_close(pg_database, AccessShareLock);
2096
2097 if (!OidIsValid(oid) && !missing_ok)
2098 ereport(ERROR,
2099 (errcode(ERRCODE_UNDEFINED_DATABASE),
2100 errmsg("database \"%s\" does not exist",
2101 dbname)));
2102
2103 return oid;
2104 }
2105
2106
2107 /*
2108 * get_database_name - given a database OID, look up the name
2109 *
2110 * Returns a palloc'd string, or NULL if no such database.
2111 */
2112 char *
2113 get_database_name(Oid dbid)
2114 {
2115 HeapTuple dbtuple;
2116 char *result;
2117
2118 dbtuple = SearchSysCache1(DATABASEOID, ObjectIdGetDatum(dbid));
2119 if (HeapTupleIsValid(dbtuple))
2120 {
2121 result = pstrdup(NameStr(((Form_pg_database) GETSTRUCT(dbtuple))->datname));
2122 ReleaseSysCache(dbtuple);
2123 }
2124 else
2125 result = NULL;
2126
2127 return result;
2128 }
2129
2130 /*
2131 * DATABASE resource manager's routines
2132 */
2133 void
2134 dbase_redo(XLogReaderState *record)
2135 {
2136 uint8 info = XLogRecGetInfo(record) & ~XLR_INFO_MASK;
2137
2138 /* Backup blocks are not used in dbase records */
2139 Assert(!XLogRecHasAnyBlockRefs(record));
2140
2141 if (info == XLOG_DBASE_CREATE)
2142 {
2143 xl_dbase_create_rec *xlrec = (xl_dbase_create_rec *) XLogRecGetData(record);
2144 char *src_path;
2145 char *dst_path;
2146 struct stat st;
2147
2148 src_path = GetDatabasePath(xlrec->src_db_id, xlrec->src_tablespace_id);
2149 dst_path = GetDatabasePath(xlrec->db_id, xlrec->tablespace_id);
2150
2151 /*
2152 * Our theory for replaying a CREATE is to forcibly drop the target
2153 * subdirectory if present, then re-copy the source data. This may be
2154 * more work than needed, but it is simple to implement.
2155 */
2156 if (stat(dst_path, &st) == 0 && S_ISDIR(st.st_mode))
2157 {
2158 if (!rmtree(dst_path, true))
2159 /* If this failed, copydir() below is going to error. */
2160 ereport(WARNING,
2161 (errmsg("some useless files may be left behind in old database directory \"%s\"",
2162 dst_path)));
2163 }
2164
2165 /*
2166 * Force dirty buffers out to disk, to ensure source database is
2167 * up-to-date for the copy.
2168 */
2169 FlushDatabaseBuffers(xlrec->src_db_id);
2170
2171 /*
2172 * Copy this subdirectory to the new location
2173 *
2174 * We don't need to copy subdirectories
2175 */
2176 copydir(src_path, dst_path, false);
2177 }
2178 else if (info == XLOG_DBASE_DROP)
2179 {
2180 xl_dbase_drop_rec *xlrec = (xl_dbase_drop_rec *) XLogRecGetData(record);
2181 char *dst_path;
2182 int i;
2183
2184 if (InHotStandby)
2185 {
2186 /*
2187 * Lock database while we resolve conflicts to ensure that
2188 * InitPostgres() cannot fully re-execute concurrently. This
2189 * avoids backends re-connecting automatically to same database,
2190 * which can happen in some cases.
2191 *
2192 * This will lock out walsenders trying to connect to db-specific
2193 * slots for logical decoding too, so it's safe for us to drop
2194 * slots.
2195 */
2196 LockSharedObjectForSession(DatabaseRelationId, xlrec->db_id, 0, AccessExclusiveLock);
2197 ResolveRecoveryConflictWithDatabase(xlrec->db_id);
2198 }
2199
2200 /* Drop any database-specific replication slots */
2201 ReplicationSlotsDropDBSlots(xlrec->db_id);
2202
2203 /* Drop pages for this database that are in the shared buffer cache */
2204 DropDatabaseBuffers(xlrec->db_id);
2205
2206 /* Also, clean out any fsync requests that might be pending in md.c */
2207 ForgetDatabaseSyncRequests(xlrec->db_id);
2208
2209 /* Clean out the xlog relcache too */
2210 XLogDropDatabase(xlrec->db_id);
2211
2212 for (i = 0; i < xlrec->ntablespaces; i++)
2213 {
2214 dst_path = GetDatabasePath(xlrec->db_id, xlrec->tablespace_ids[i]);
2215
2216 /* And remove the physical files */
2217 if (!rmtree(dst_path, true))
2218 ereport(WARNING,
2219 (errmsg("some useless files may be left behind in old database directory \"%s\"",
2220 dst_path)));
2221 pfree(dst_path);
2222 }
2223
2224 if (InHotStandby)
2225 {
2226 /*
2227 * Release locks prior to commit. XXX There is a race condition
2228 * here that may allow backends to reconnect, but the window for
2229 * this is small because the gap between here and commit is mostly
2230 * fairly small and it is unlikely that people will be dropping
2231 * databases that we are trying to connect to anyway.
2232 */
2233 UnlockSharedObjectForSession(DatabaseRelationId, xlrec->db_id, 0, AccessExclusiveLock);
2234 }
2235 }
2236 else
2237 elog(PANIC, "dbase_redo: unknown op code %u", info);
2238 }
Postgresql Clone from git://git.postgresql.org/git/postgresql.git