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