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