Branch data Line data Source code
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-2026, PostgreSQL Global Development Group
12 : : * Portions Copyright (c) 1994, Regents of the University of California
13 : : *
14 : : *
15 : : * IDENTIFICATION
16 : : * src/backend/commands/dbcommands.c
17 : : *
18 : : *-------------------------------------------------------------------------
19 : : */
20 : : #include "postgres.h"
21 : :
22 : : #include <fcntl.h>
23 : : #include <unistd.h>
24 : : #include <sys/stat.h>
25 : :
26 : : #include "access/genam.h"
27 : : #include "access/heapam.h"
28 : : #include "access/htup_details.h"
29 : : #include "access/multixact.h"
30 : : #include "access/tableam.h"
31 : : #include "access/xact.h"
32 : : #include "access/xloginsert.h"
33 : : #include "access/xlogrecovery.h"
34 : : #include "access/xlogutils.h"
35 : : #include "catalog/catalog.h"
36 : : #include "catalog/dependency.h"
37 : : #include "catalog/indexing.h"
38 : : #include "catalog/objectaccess.h"
39 : : #include "catalog/pg_authid.h"
40 : : #include "catalog/pg_collation.h"
41 : : #include "catalog/pg_database.h"
42 : : #include "catalog/pg_db_role_setting.h"
43 : : #include "catalog/pg_subscription.h"
44 : : #include "catalog/pg_tablespace.h"
45 : : #include "commands/comment.h"
46 : : #include "commands/dbcommands.h"
47 : : #include "commands/dbcommands_xlog.h"
48 : : #include "commands/defrem.h"
49 : : #include "commands/seclabel.h"
50 : : #include "commands/tablespace.h"
51 : : #include "common/file_perm.h"
52 : : #include "mb/pg_wchar.h"
53 : : #include "miscadmin.h"
54 : : #include "pgstat.h"
55 : : #include "postmaster/bgwriter.h"
56 : : #include "replication/slot.h"
57 : : #include "storage/copydir.h"
58 : : #include "storage/fd.h"
59 : : #include "storage/ipc.h"
60 : : #include "storage/lmgr.h"
61 : : #include "storage/md.h"
62 : : #include "storage/procarray.h"
63 : : #include "storage/procsignal.h"
64 : : #include "storage/smgr.h"
65 : : #include "utils/acl.h"
66 : : #include "utils/builtins.h"
67 : : #include "utils/fmgroids.h"
68 : : #include "utils/injection_point.h"
69 : : #include "utils/lsyscache.h"
70 : : #include "utils/pg_locale.h"
71 : : #include "utils/relmapper.h"
72 : : #include "utils/snapmgr.h"
73 : : #include "utils/syscache.h"
74 : : #include "utils/wait_event.h"
75 : :
76 : : /*
77 : : * Create database strategy.
78 : : *
79 : : * CREATEDB_WAL_LOG will copy the database at the block level and WAL log each
80 : : * copied block.
81 : : *
82 : : * CREATEDB_FILE_COPY will simply perform a file system level copy of the
83 : : * database and log a single record for each tablespace copied. To make this
84 : : * safe, it also triggers checkpoints before and after the operation.
85 : : */
86 : : typedef enum CreateDBStrategy
87 : : {
88 : : CREATEDB_WAL_LOG,
89 : : CREATEDB_FILE_COPY,
90 : : } CreateDBStrategy;
91 : :
92 : : typedef struct
93 : : {
94 : : Oid src_dboid; /* source (template) DB */
95 : : Oid dest_dboid; /* DB we are trying to create */
96 : : CreateDBStrategy strategy; /* create db strategy */
97 : : } createdb_failure_params;
98 : :
99 : : typedef struct
100 : : {
101 : : Oid dest_dboid; /* DB we are trying to move */
102 : : Oid dest_tsoid; /* tablespace we are trying to move to */
103 : : } movedb_failure_params;
104 : :
105 : : /*
106 : : * Information about a relation to be copied when creating a database.
107 : : */
108 : : typedef struct CreateDBRelInfo
109 : : {
110 : : RelFileLocator rlocator; /* physical relation identifier */
111 : : Oid reloid; /* relation oid */
112 : : bool permanent; /* relation is permanent or unlogged */
113 : : } CreateDBRelInfo;
114 : :
115 : :
116 : : /* non-export function prototypes */
117 : : static void createdb_failure_callback(int code, Datum arg);
118 : : static void movedb(const char *dbname, const char *tblspcname);
119 : : static void movedb_failure_callback(int code, Datum arg);
120 : : static bool get_db_info(const char *name, LOCKMODE lockmode,
121 : : Oid *dbIdP, Oid *ownerIdP,
122 : : int *encodingP, bool *dbIsTemplateP, bool *dbAllowConnP, bool *dbHasLoginEvtP,
123 : : TransactionId *dbFrozenXidP, MultiXactId *dbMinMultiP,
124 : : Oid *dbTablespace, char **dbCollate, char **dbCtype, char **dbLocale,
125 : : char **dbIcurules,
126 : : char *dbLocProvider,
127 : : char **dbCollversion);
128 : : static void remove_dbtablespaces(Oid db_id);
129 : : static bool check_db_file_conflict(Oid db_id);
130 : : static int errdetail_busy_db(int notherbackends, int npreparedxacts);
131 : : static void CreateDatabaseUsingWalLog(Oid src_dboid, Oid dst_dboid, Oid src_tsid,
132 : : Oid dst_tsid);
133 : : static List *ScanSourceDatabasePgClass(Oid tbid, Oid dbid, char *srcpath);
134 : : static List *ScanSourceDatabasePgClassPage(Page page, Buffer buf, Oid tbid,
135 : : Oid dbid, char *srcpath,
136 : : List *rlocatorlist, Snapshot snapshot);
137 : : static CreateDBRelInfo *ScanSourceDatabasePgClassTuple(HeapTupleData *tuple,
138 : : Oid tbid, Oid dbid,
139 : : char *srcpath);
140 : : static void CreateDirAndVersionFile(char *dbpath, Oid dbid, Oid tsid,
141 : : bool isRedo);
142 : : static void CreateDatabaseUsingFileCopy(Oid src_dboid, Oid dst_dboid,
143 : : Oid src_tsid, Oid dst_tsid);
144 : : static void recovery_create_dbdir(char *path, bool only_tblspc);
145 : :
146 : : /*
147 : : * Create a new database using the WAL_LOG strategy.
148 : : *
149 : : * Each copied block is separately written to the write-ahead log.
150 : : */
151 : : static void
152 : 273 : CreateDatabaseUsingWalLog(Oid src_dboid, Oid dst_dboid,
153 : : Oid src_tsid, Oid dst_tsid)
154 : : {
155 : : char *srcpath;
156 : : char *dstpath;
157 : 273 : List *rlocatorlist = NULL;
158 : : ListCell *cell;
159 : : LockRelId srcrelid;
160 : : LockRelId dstrelid;
161 : : RelFileLocator srcrlocator;
162 : : RelFileLocator dstrlocator;
163 : : CreateDBRelInfo *relinfo;
164 : :
165 : : /* Get source and destination database paths. */
166 : 273 : srcpath = GetDatabasePath(src_dboid, src_tsid);
167 : 273 : dstpath = GetDatabasePath(dst_dboid, dst_tsid);
168 : :
169 : : /* Create database directory and write PG_VERSION file. */
170 : 273 : CreateDirAndVersionFile(dstpath, dst_dboid, dst_tsid, false);
171 : :
172 : : /* Copy relmap file from source database to the destination database. */
173 : 273 : RelationMapCopy(dst_dboid, dst_tsid, srcpath, dstpath);
174 : :
175 : : /* Get list of relfilelocators to copy from the source database. */
176 : 273 : rlocatorlist = ScanSourceDatabasePgClass(src_tsid, src_dboid, srcpath);
177 : : Assert(rlocatorlist != NIL);
178 : :
179 : : /*
180 : : * Database IDs will be the same for all relations so set them before
181 : : * entering the loop.
182 : : */
183 : 273 : srcrelid.dbId = src_dboid;
184 : 273 : dstrelid.dbId = dst_dboid;
185 : :
186 : : /* Loop over our list of relfilelocators and copy each one. */
187 [ + - + + : 66422 : foreach(cell, rlocatorlist)
+ + ]
188 : : {
189 : 66151 : relinfo = lfirst(cell);
190 : 66151 : srcrlocator = relinfo->rlocator;
191 : :
192 : : /*
193 : : * If the relation is from the source db's default tablespace then we
194 : : * need to create it in the destination db's default tablespace.
195 : : * Otherwise, we need to create in the same tablespace as it is in the
196 : : * source database.
197 : : */
198 [ + - ]: 66151 : if (srcrlocator.spcOid == src_tsid)
199 : 66151 : dstrlocator.spcOid = dst_tsid;
200 : : else
201 : 0 : dstrlocator.spcOid = srcrlocator.spcOid;
202 : :
203 : 66151 : dstrlocator.dbOid = dst_dboid;
204 : 66151 : dstrlocator.relNumber = srcrlocator.relNumber;
205 : :
206 : : /*
207 : : * Acquire locks on source and target relations before copying.
208 : : *
209 : : * We typically do not read relation data into shared_buffers without
210 : : * holding a relation lock. It's unclear what could go wrong if we
211 : : * skipped it in this case, because nobody can be modifying either the
212 : : * source or destination database at this point, and we have locks on
213 : : * both databases, too, but let's take the conservative route.
214 : : */
215 : 66151 : dstrelid.relId = srcrelid.relId = relinfo->reloid;
216 : 66151 : LockRelationId(&srcrelid, AccessShareLock);
217 : 66151 : LockRelationId(&dstrelid, AccessShareLock);
218 : :
219 : : /* Copy relation storage from source to the destination. */
220 : 66151 : CreateAndCopyRelationData(srcrlocator, dstrlocator, relinfo->permanent);
221 : :
222 : : /* Release the relation locks. */
223 : 66149 : UnlockRelationId(&srcrelid, AccessShareLock);
224 : 66149 : UnlockRelationId(&dstrelid, AccessShareLock);
225 : : }
226 : :
227 : 271 : pfree(srcpath);
228 : 271 : pfree(dstpath);
229 : 271 : list_free_deep(rlocatorlist);
230 : 271 : }
231 : :
232 : : /*
233 : : * Scan the pg_class table in the source database to identify the relations
234 : : * that need to be copied to the destination database.
235 : : *
236 : : * This is an exception to the usual rule that cross-database access is
237 : : * not possible. We can make it work here because we know that there are no
238 : : * connections to the source database and (since there can't be prepared
239 : : * transactions touching that database) no in-doubt tuples either. This
240 : : * means that we don't need to worry about pruning removing anything from
241 : : * under us, and we don't need to be too picky about our snapshot either.
242 : : * As long as it sees all previously-committed XIDs as committed and all
243 : : * aborted XIDs as aborted, we should be fine: nothing else is possible
244 : : * here.
245 : : *
246 : : * We can't rely on the relcache for anything here, because that only knows
247 : : * about the database to which we are connected, and can't handle access to
248 : : * other databases. That also means we can't rely on the heap scan
249 : : * infrastructure, which would be a bad idea anyway since it might try
250 : : * to do things like HOT pruning which we definitely can't do safely in
251 : : * a database to which we're not even connected.
252 : : */
253 : : static List *
254 : 273 : ScanSourceDatabasePgClass(Oid tbid, Oid dbid, char *srcpath)
255 : : {
256 : : RelFileLocator rlocator;
257 : : BlockNumber nblocks;
258 : : BlockNumber blkno;
259 : : Buffer buf;
260 : : RelFileNumber relfilenumber;
261 : : Page page;
262 : 273 : List *rlocatorlist = NIL;
263 : : LockRelId relid;
264 : : Snapshot snapshot;
265 : : SMgrRelation smgr;
266 : : BufferAccessStrategy bstrategy;
267 : :
268 : : /* Get pg_class relfilenumber. */
269 : 273 : relfilenumber = RelationMapOidToFilenumberForDatabase(srcpath,
270 : : RelationRelationId);
271 : :
272 : : /* Don't read data into shared_buffers without holding a relation lock. */
273 : 273 : relid.dbId = dbid;
274 : 273 : relid.relId = RelationRelationId;
275 : 273 : LockRelationId(&relid, AccessShareLock);
276 : :
277 : : /* Prepare a RelFileLocator for the pg_class relation. */
278 : 273 : rlocator.spcOid = tbid;
279 : 273 : rlocator.dbOid = dbid;
280 : 273 : rlocator.relNumber = relfilenumber;
281 : :
282 : 273 : smgr = smgropen(rlocator, INVALID_PROC_NUMBER);
283 : 273 : nblocks = smgrnblocks(smgr, MAIN_FORKNUM);
284 : 273 : smgrclose(smgr);
285 : :
286 : : /* Use a buffer access strategy since this is a bulk read operation. */
287 : 273 : bstrategy = GetAccessStrategy(BAS_BULKREAD);
288 : :
289 : : /*
290 : : * As explained in the function header comments, we need a snapshot that
291 : : * will see all committed transactions as committed, and our transaction
292 : : * snapshot - or the active snapshot - might not be new enough for that,
293 : : * but the return value of GetLatestSnapshot() should work fine.
294 : : */
295 : 273 : snapshot = RegisterSnapshot(GetLatestSnapshot());
296 : :
297 : : /* Process the relation block by block. */
298 [ + + ]: 4641 : for (blkno = 0; blkno < nblocks; blkno++)
299 : : {
300 [ - + ]: 4368 : CHECK_FOR_INTERRUPTS();
301 : :
302 : 4368 : buf = ReadBufferWithoutRelcache(rlocator, MAIN_FORKNUM, blkno,
303 : : RBM_NORMAL, bstrategy, true);
304 : :
305 : 4368 : LockBuffer(buf, BUFFER_LOCK_SHARE);
306 : 4368 : page = BufferGetPage(buf);
307 [ + - - + ]: 4368 : if (PageIsNew(page) || PageIsEmpty(page))
308 : : {
309 : 0 : UnlockReleaseBuffer(buf);
310 : 0 : continue;
311 : : }
312 : :
313 : : /* Append relevant pg_class tuples for current page to rlocatorlist. */
314 : 4368 : rlocatorlist = ScanSourceDatabasePgClassPage(page, buf, tbid, dbid,
315 : : srcpath, rlocatorlist,
316 : : snapshot);
317 : :
318 : 4368 : UnlockReleaseBuffer(buf);
319 : : }
320 : 273 : UnregisterSnapshot(snapshot);
321 : :
322 : : /* Release relation lock. */
323 : 273 : UnlockRelationId(&relid, AccessShareLock);
324 : :
325 : 273 : return rlocatorlist;
326 : : }
327 : :
328 : : /*
329 : : * Scan one page of the source database's pg_class relation and add relevant
330 : : * entries to rlocatorlist. The return value is the updated list.
331 : : */
332 : : static List *
333 : 4368 : ScanSourceDatabasePgClassPage(Page page, Buffer buf, Oid tbid, Oid dbid,
334 : : char *srcpath, List *rlocatorlist,
335 : : Snapshot snapshot)
336 : : {
337 : 4368 : BlockNumber blkno = BufferGetBlockNumber(buf);
338 : : OffsetNumber offnum;
339 : : OffsetNumber maxoff;
340 : : HeapTupleData tuple;
341 : :
342 : 4368 : maxoff = PageGetMaxOffsetNumber(page);
343 : :
344 : : /* Loop over offsets. */
345 : 4368 : for (offnum = FirstOffsetNumber;
346 [ + + ]: 196393 : offnum <= maxoff;
347 : 192025 : offnum = OffsetNumberNext(offnum))
348 : : {
349 : : ItemId itemid;
350 : :
351 : 192025 : itemid = PageGetItemId(page, offnum);
352 : :
353 : : /* Nothing to do if slot is empty or already dead. */
354 [ + + + + ]: 192025 : if (!ItemIdIsUsed(itemid) || ItemIdIsDead(itemid) ||
355 [ + + ]: 153475 : ItemIdIsRedirected(itemid))
356 : 68307 : continue;
357 : :
358 : : Assert(ItemIdIsNormal(itemid));
359 : 123718 : ItemPointerSet(&(tuple.t_self), blkno, offnum);
360 : :
361 : : /* Initialize a HeapTupleData structure. */
362 : 123718 : tuple.t_data = (HeapTupleHeader) PageGetItem(page, itemid);
363 : 123718 : tuple.t_len = ItemIdGetLength(itemid);
364 : 123718 : tuple.t_tableOid = RelationRelationId;
365 : :
366 : : /* Skip tuples that are not visible to this snapshot. */
367 [ + + ]: 123718 : if (HeapTupleSatisfiesVisibility(&tuple, snapshot, buf))
368 : : {
369 : : CreateDBRelInfo *relinfo;
370 : :
371 : : /*
372 : : * ScanSourceDatabasePgClassTuple is in charge of constructing a
373 : : * CreateDBRelInfo object for this tuple, but can also decide that
374 : : * this tuple isn't something we need to copy. If we do need to
375 : : * copy the relation, add it to the list.
376 : : */
377 : 123696 : relinfo = ScanSourceDatabasePgClassTuple(&tuple, tbid, dbid,
378 : : srcpath);
379 [ + + ]: 123696 : if (relinfo != NULL)
380 : 66639 : rlocatorlist = lappend(rlocatorlist, relinfo);
381 : : }
382 : : }
383 : :
384 : 4368 : return rlocatorlist;
385 : : }
386 : :
387 : : /*
388 : : * Decide whether a certain pg_class tuple represents something that
389 : : * needs to be copied from the source database to the destination database,
390 : : * and if so, construct a CreateDBRelInfo for it.
391 : : *
392 : : * Visibility checks are handled by the caller, so our job here is just
393 : : * to assess the data stored in the tuple.
394 : : */
395 : : CreateDBRelInfo *
396 : 123696 : ScanSourceDatabasePgClassTuple(HeapTupleData *tuple, Oid tbid, Oid dbid,
397 : : char *srcpath)
398 : : {
399 : : CreateDBRelInfo *relinfo;
400 : : Form_pg_class classForm;
401 : 123696 : RelFileNumber relfilenumber = InvalidRelFileNumber;
402 : :
403 : 123696 : classForm = (Form_pg_class) GETSTRUCT(tuple);
404 : :
405 : : /*
406 : : * Return NULL if this object does not need to be copied.
407 : : *
408 : : * Shared objects don't need to be copied, because they are shared.
409 : : * Objects without storage can't be copied, because there's nothing to
410 : : * copy. Temporary relations don't need to be copied either, because they
411 : : * are inaccessible outside of the session that created them, which must
412 : : * be gone already, and couldn't connect to a different database if it
413 : : * still existed. autovacuum will eventually remove the pg_class entries
414 : : * as well.
415 : : */
416 [ + + ]: 123696 : if (classForm->reltablespace == GLOBALTABLESPACE_OID ||
417 [ + + + + : 111138 : !RELKIND_HAS_STORAGE(classForm->relkind) ||
+ + + + -
+ ]
418 [ - + ]: 66639 : classForm->relpersistence == RELPERSISTENCE_TEMP)
419 : 57057 : return NULL;
420 : :
421 : : /*
422 : : * If relfilenumber is valid then directly use it. Otherwise, consult the
423 : : * relmap.
424 : : */
425 [ + + ]: 66639 : if (RelFileNumberIsValid(classForm->relfilenode))
426 : 61998 : relfilenumber = classForm->relfilenode;
427 : : else
428 : 4641 : relfilenumber = RelationMapOidToFilenumberForDatabase(srcpath,
429 : : classForm->oid);
430 : :
431 : : /* We must have a valid relfilenumber. */
432 [ - + ]: 66639 : if (!RelFileNumberIsValid(relfilenumber))
433 [ # # ]: 0 : elog(ERROR, "relation with OID %u does not have a valid relfilenumber",
434 : : classForm->oid);
435 : :
436 : : /* Prepare a rel info element and add it to the list. */
437 : 66639 : relinfo = palloc_object(CreateDBRelInfo);
438 [ - + ]: 66639 : if (OidIsValid(classForm->reltablespace))
439 : 0 : relinfo->rlocator.spcOid = classForm->reltablespace;
440 : : else
441 : 66639 : relinfo->rlocator.spcOid = tbid;
442 : :
443 : 66639 : relinfo->rlocator.dbOid = dbid;
444 : 66639 : relinfo->rlocator.relNumber = relfilenumber;
445 : 66639 : relinfo->reloid = classForm->oid;
446 : :
447 : : /* Temporary relations were rejected above. */
448 : : Assert(classForm->relpersistence != RELPERSISTENCE_TEMP);
449 : 66639 : relinfo->permanent =
450 : 66639 : (classForm->relpersistence == RELPERSISTENCE_PERMANENT) ? true : false;
451 : :
452 : 66639 : return relinfo;
453 : : }
454 : :
455 : : /*
456 : : * Create database directory and write out the PG_VERSION file in the database
457 : : * path. If isRedo is true, it's okay for the database directory to exist
458 : : * already.
459 : : */
460 : : static void
461 : 297 : CreateDirAndVersionFile(char *dbpath, Oid dbid, Oid tsid, bool isRedo)
462 : : {
463 : : int fd;
464 : : size_t nbytes;
465 : : char versionfile[MAXPGPATH];
466 : : char buf[16];
467 : :
468 : : /*
469 : : * Note that we don't have to copy version data from the source database;
470 : : * there's only one legal value.
471 : : */
472 : 297 : sprintf(buf, "%s\n", PG_MAJORVERSION);
473 : 297 : nbytes = strlen(PG_MAJORVERSION) + 1;
474 : :
475 : : /* Create database directory. */
476 [ + + ]: 297 : if (MakePGDirectory(dbpath) < 0)
477 : : {
478 : : /* Failure other than already exists or not in WAL replay? */
479 [ + - - + ]: 9 : if (errno != EEXIST || !isRedo)
480 [ # # ]: 0 : ereport(ERROR,
481 : : (errcode_for_file_access(),
482 : : errmsg("could not create directory \"%s\": %m", dbpath)));
483 : : }
484 : :
485 : : /*
486 : : * Create PG_VERSION file in the database path. If the file already
487 : : * exists and we are in WAL replay then try again to open it in write
488 : : * mode.
489 : : */
490 : 297 : snprintf(versionfile, sizeof(versionfile), "%s/%s", dbpath, "PG_VERSION");
491 : :
492 : 297 : fd = OpenTransientFile(versionfile, O_WRONLY | O_CREAT | O_EXCL | PG_BINARY);
493 [ + + + - : 297 : if (fd < 0 && errno == EEXIST && isRedo)
+ - ]
494 : 9 : fd = OpenTransientFile(versionfile, O_WRONLY | O_TRUNC | PG_BINARY);
495 : :
496 [ - + ]: 297 : if (fd < 0)
497 [ # # ]: 0 : ereport(ERROR,
498 : : (errcode_for_file_access(),
499 : : errmsg("could not create file \"%s\": %m", versionfile)));
500 : :
501 : : /* Write PG_MAJORVERSION in the PG_VERSION file. */
502 : 297 : pgstat_report_wait_start(WAIT_EVENT_VERSION_FILE_WRITE);
503 : 297 : errno = 0;
504 [ - + ]: 297 : if (write(fd, buf, nbytes) != nbytes)
505 : : {
506 : : /* If write didn't set errno, assume problem is no disk space. */
507 [ # # ]: 0 : if (errno == 0)
508 : 0 : errno = ENOSPC;
509 [ # # ]: 0 : ereport(ERROR,
510 : : (errcode_for_file_access(),
511 : : errmsg("could not write to file \"%s\": %m", versionfile)));
512 : : }
513 : 297 : pgstat_report_wait_end();
514 : :
515 : 297 : pgstat_report_wait_start(WAIT_EVENT_VERSION_FILE_SYNC);
516 [ - + ]: 297 : if (pg_fsync(fd) != 0)
517 [ # # ]: 0 : ereport(data_sync_elevel(ERROR),
518 : : (errcode_for_file_access(),
519 : : errmsg("could not fsync file \"%s\": %m", versionfile)));
520 : 297 : fsync_fname(dbpath, true);
521 : 297 : pgstat_report_wait_end();
522 : :
523 : : /* Close the version file. */
524 : 297 : CloseTransientFile(fd);
525 : :
526 : : /* If we are not in WAL replay then write the WAL. */
527 [ + + ]: 297 : if (!isRedo)
528 : : {
529 : : xl_dbase_create_wal_log_rec xlrec;
530 : :
531 : 273 : START_CRIT_SECTION();
532 : :
533 : 273 : xlrec.db_id = dbid;
534 : 273 : xlrec.tablespace_id = tsid;
535 : :
536 : 273 : XLogBeginInsert();
537 : 273 : XLogRegisterData(&xlrec,
538 : : sizeof(xl_dbase_create_wal_log_rec));
539 : :
540 : 273 : (void) XLogInsert(RM_DBASE_ID, XLOG_DBASE_CREATE_WAL_LOG);
541 : :
542 : 273 : END_CRIT_SECTION();
543 : : }
544 : 297 : }
545 : :
546 : : /*
547 : : * Create a new database using the FILE_COPY strategy.
548 : : *
549 : : * Copy each tablespace at the filesystem level, and log a single WAL record
550 : : * for each tablespace copied. This requires a checkpoint before and after the
551 : : * copy, which may be expensive, but it does greatly reduce WAL generation
552 : : * if the copied database is large.
553 : : */
554 : : static void
555 : 153 : CreateDatabaseUsingFileCopy(Oid src_dboid, Oid dst_dboid, Oid src_tsid,
556 : : Oid dst_tsid)
557 : : {
558 : : TableScanDesc scan;
559 : : Relation rel;
560 : : HeapTuple tuple;
561 : :
562 : : /*
563 : : * The strategy check in createdb() runs before our transaction has an XID
564 : : * and before the pg_database row exists, so the datachecksumsworker
565 : : * launcher can start in that window and miss both the new database and
566 : : * our transaction, leaving the raw-copied files without checksums.
567 : : */
568 [ + + ]: 153 : if (DataChecksumsInProgressOn())
569 [ + - ]: 1 : ereport(ERROR,
570 : : errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
571 : : errmsg("create database strategy \"%s\" not allowed when data checksums are being enabled",
572 : : "file_copy"));
573 : :
574 : : /*
575 : : * The XID is assigned by now, so a datachecksumsworker launcher starting
576 : : * after this point will wait for us and find the new database.
577 : : */
578 : :
579 : : /*
580 : : * Force a checkpoint before starting the copy. This will force all dirty
581 : : * buffers, including those of unlogged tables, out to disk, to ensure
582 : : * source database is up-to-date on disk for the copy.
583 : : * FlushDatabaseBuffers() would suffice for that, but we also want to
584 : : * process any pending unlink requests. Otherwise, if a checkpoint
585 : : * happened while we're copying files, a file might be deleted just when
586 : : * we're about to copy it, causing the lstat() call in copydir() to fail
587 : : * with ENOENT.
588 : : *
589 : : * In binary upgrade mode, we can skip this checkpoint because pg_upgrade
590 : : * is careful to ensure that template0 is fully written to disk prior to
591 : : * any CREATE DATABASE commands.
592 : : */
593 [ + + ]: 152 : if (!IsBinaryUpgrade)
594 : 120 : RequestCheckpoint(CHECKPOINT_FAST | CHECKPOINT_FORCE |
595 : : CHECKPOINT_WAIT | CHECKPOINT_FLUSH_UNLOGGED);
596 : :
597 : : /*
598 : : * Iterate through all tablespaces of the template database, and copy each
599 : : * one to the new database.
600 : : */
601 : 152 : rel = table_open(TableSpaceRelationId, AccessShareLock);
602 : 152 : scan = table_beginscan_catalog(rel, 0, NULL);
603 [ + + ]: 490 : while ((tuple = heap_getnext(scan, ForwardScanDirection)) != NULL)
604 : : {
605 : 338 : Form_pg_tablespace spaceform = (Form_pg_tablespace) GETSTRUCT(tuple);
606 : 338 : Oid srctablespace = spaceform->oid;
607 : : Oid dsttablespace;
608 : : char *srcpath;
609 : : char *dstpath;
610 : : struct stat st;
611 : :
612 : : /* No need to copy global tablespace */
613 [ + + ]: 338 : if (srctablespace == GLOBALTABLESPACE_OID)
614 : 186 : continue;
615 : :
616 : 186 : srcpath = GetDatabasePath(src_dboid, srctablespace);
617 : :
618 [ + + + - : 338 : if (stat(srcpath, &st) < 0 || !S_ISDIR(st.st_mode) ||
- + ]
619 : 152 : directory_is_empty(srcpath))
620 : : {
621 : : /* Assume we can ignore it */
622 : 34 : pfree(srcpath);
623 : 34 : continue;
624 : : }
625 : :
626 [ + - ]: 152 : if (srctablespace == src_tsid)
627 : 152 : dsttablespace = dst_tsid;
628 : : else
629 : 0 : dsttablespace = srctablespace;
630 : :
631 : 152 : dstpath = GetDatabasePath(dst_dboid, dsttablespace);
632 : :
633 : : /*
634 : : * Copy this subdirectory to the new location
635 : : *
636 : : * We don't need to copy subdirectories
637 : : */
638 : 152 : copydir(srcpath, dstpath, false);
639 : :
640 : : /* Record the filesystem change in XLOG */
641 : : {
642 : : xl_dbase_create_file_copy_rec xlrec;
643 : :
644 : 152 : xlrec.db_id = dst_dboid;
645 : 152 : xlrec.tablespace_id = dsttablespace;
646 : 152 : xlrec.src_db_id = src_dboid;
647 : 152 : xlrec.src_tablespace_id = srctablespace;
648 : :
649 : 152 : XLogBeginInsert();
650 : 152 : XLogRegisterData(&xlrec,
651 : : sizeof(xl_dbase_create_file_copy_rec));
652 : :
653 : 152 : (void) XLogInsert(RM_DBASE_ID,
654 : : XLOG_DBASE_CREATE_FILE_COPY | XLR_SPECIAL_REL_UPDATE);
655 : : }
656 : 152 : pfree(srcpath);
657 : 152 : pfree(dstpath);
658 : : }
659 : 152 : table_endscan(scan);
660 : 152 : table_close(rel, AccessShareLock);
661 : :
662 : : /*
663 : : * We force a checkpoint before committing. This effectively means that
664 : : * committed XLOG_DBASE_CREATE_FILE_COPY operations will never need to be
665 : : * replayed (at least not in ordinary crash recovery; we still have to
666 : : * make the XLOG entry for the benefit of PITR operations). This avoids
667 : : * two nasty scenarios:
668 : : *
669 : : * #1: At wal_level=minimal, we don't XLOG the contents of newly created
670 : : * relfilenodes; therefore the drop-and-recreate-whole-directory behavior
671 : : * of DBASE_CREATE replay would lose such files created in the new
672 : : * database between our commit and the next checkpoint.
673 : : *
674 : : * #2: Since we have to recopy the source database during DBASE_CREATE
675 : : * replay, we run the risk of copying changes in it that were committed
676 : : * after the original CREATE DATABASE command but before the system crash
677 : : * that led to the replay. This is at least unexpected and at worst could
678 : : * lead to inconsistencies, eg duplicate table names.
679 : : *
680 : : * (Both of these were real bugs in releases 8.0 through 8.0.3.)
681 : : *
682 : : * In PITR replay, the first of these isn't an issue, and the second is
683 : : * only a risk if the CREATE DATABASE and subsequent template database
684 : : * change both occur while a base backup is being taken. There doesn't
685 : : * seem to be much we can do about that except document it as a
686 : : * limitation.
687 : : *
688 : : * In binary upgrade mode, we can skip this checkpoint because neither of
689 : : * these problems applies: we don't ever replay the WAL generated during
690 : : * pg_upgrade, and we don't support taking base backups during pg_upgrade
691 : : * (not to mention that we don't concurrently modify template0, either).
692 : : *
693 : : * See CreateDatabaseUsingWalLog() for a less cheesy CREATE DATABASE
694 : : * strategy that avoids these problems.
695 : : */
696 [ + + ]: 152 : if (!IsBinaryUpgrade)
697 : 120 : RequestCheckpoint(CHECKPOINT_FAST | CHECKPOINT_FORCE |
698 : : CHECKPOINT_WAIT);
699 : 152 : }
700 : :
701 : : /*
702 : : * CREATE DATABASE
703 : : */
704 : : Oid
705 : 448 : createdb(ParseState *pstate, const CreatedbStmt *stmt)
706 : : {
707 : : Oid src_dboid;
708 : : Oid src_owner;
709 : 448 : int src_encoding = -1;
710 : 448 : char *src_collate = NULL;
711 : 448 : char *src_ctype = NULL;
712 : 448 : char *src_locale = NULL;
713 : 448 : char *src_icurules = NULL;
714 : 448 : char src_locprovider = '\0';
715 : 448 : char *src_collversion = NULL;
716 : : bool src_istemplate;
717 : 448 : bool src_hasloginevt = false;
718 : : bool src_allowconn;
719 : 448 : TransactionId src_frozenxid = InvalidTransactionId;
720 : 448 : MultiXactId src_minmxid = InvalidMultiXactId;
721 : : Oid src_deftablespace;
722 : : volatile Oid dst_deftablespace;
723 : : Relation pg_database_rel;
724 : : HeapTuple tuple;
725 : 448 : Datum new_record[Natts_pg_database] = {0};
726 : 448 : bool new_record_nulls[Natts_pg_database] = {0};
727 : 448 : Oid dboid = InvalidOid;
728 : : Oid datdba;
729 : : ListCell *option;
730 : 448 : DefElem *tablespacenameEl = NULL;
731 : 448 : DefElem *ownerEl = NULL;
732 : 448 : DefElem *templateEl = NULL;
733 : 448 : DefElem *encodingEl = NULL;
734 : 448 : DefElem *localeEl = NULL;
735 : 448 : DefElem *builtinlocaleEl = NULL;
736 : 448 : DefElem *collateEl = NULL;
737 : 448 : DefElem *ctypeEl = NULL;
738 : 448 : DefElem *iculocaleEl = NULL;
739 : 448 : DefElem *icurulesEl = NULL;
740 : 448 : DefElem *locproviderEl = NULL;
741 : 448 : DefElem *istemplateEl = NULL;
742 : 448 : DefElem *allowconnectionsEl = NULL;
743 : 448 : DefElem *connlimitEl = NULL;
744 : 448 : DefElem *collversionEl = NULL;
745 : 448 : DefElem *strategyEl = NULL;
746 : 448 : char *dbname = stmt->dbname;
747 : 448 : char *dbowner = NULL;
748 : 448 : const char *dbtemplate = NULL;
749 : 448 : char *dbcollate = NULL;
750 : 448 : char *dbctype = NULL;
751 : 448 : const char *dblocale = NULL;
752 : 448 : char *dbicurules = NULL;
753 : 448 : char dblocprovider = '\0';
754 : : char *canonname;
755 : 448 : int encoding = -1;
756 : 448 : bool dbistemplate = false;
757 : 448 : bool dballowconnections = true;
758 : 448 : int dbconnlimit = DATCONNLIMIT_UNLIMITED;
759 : 448 : char *dbcollversion = NULL;
760 : : int notherbackends;
761 : : int npreparedxacts;
762 : 448 : CreateDBStrategy dbstrategy = CREATEDB_WAL_LOG;
763 : : createdb_failure_params fparms;
764 : :
765 : : /* Report error if name has \n or \r character. */
766 [ + + ]: 448 : if (strpbrk(dbname, "\n\r"))
767 [ + - ]: 3 : ereport(ERROR,
768 : : (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
769 : : errmsg("database name \"%s\" contains a newline or carriage return character", dbname)));
770 : :
771 : : /* Extract options from the statement node tree */
772 [ + + + + : 1329 : foreach(option, stmt->options)
+ + ]
773 : : {
774 : 884 : DefElem *defel = (DefElem *) lfirst(option);
775 : :
776 [ + + ]: 884 : if (strcmp(defel->defname, "tablespace") == 0)
777 : : {
778 [ - + ]: 17 : if (tablespacenameEl)
779 : 0 : errorConflictingDefElem(defel, pstate);
780 : 17 : tablespacenameEl = defel;
781 : : }
782 [ - + ]: 867 : else if (strcmp(defel->defname, "owner") == 0)
783 : : {
784 [ # # ]: 0 : if (ownerEl)
785 : 0 : errorConflictingDefElem(defel, pstate);
786 : 0 : ownerEl = defel;
787 : : }
788 [ + + ]: 867 : else if (strcmp(defel->defname, "template") == 0)
789 : : {
790 [ - + ]: 197 : if (templateEl)
791 : 0 : errorConflictingDefElem(defel, pstate);
792 : 197 : templateEl = defel;
793 : : }
794 [ + + ]: 670 : else if (strcmp(defel->defname, "encoding") == 0)
795 : : {
796 [ - + ]: 58 : if (encodingEl)
797 : 0 : errorConflictingDefElem(defel, pstate);
798 : 58 : encodingEl = defel;
799 : : }
800 [ + + ]: 612 : else if (strcmp(defel->defname, "locale") == 0)
801 : : {
802 [ - + ]: 57 : if (localeEl)
803 : 0 : errorConflictingDefElem(defel, pstate);
804 : 57 : localeEl = defel;
805 : : }
806 [ + + ]: 555 : else if (strcmp(defel->defname, "builtin_locale") == 0)
807 : : {
808 [ - + ]: 8 : if (builtinlocaleEl)
809 : 0 : errorConflictingDefElem(defel, pstate);
810 : 8 : builtinlocaleEl = defel;
811 : : }
812 [ + + ]: 547 : else if (strcmp(defel->defname, "lc_collate") == 0)
813 : : {
814 [ - + ]: 11 : if (collateEl)
815 : 0 : errorConflictingDefElem(defel, pstate);
816 : 11 : collateEl = defel;
817 : : }
818 [ + + ]: 536 : else if (strcmp(defel->defname, "lc_ctype") == 0)
819 : : {
820 [ - + ]: 11 : if (ctypeEl)
821 : 0 : errorConflictingDefElem(defel, pstate);
822 : 11 : ctypeEl = defel;
823 : : }
824 [ + + ]: 525 : else if (strcmp(defel->defname, "icu_locale") == 0)
825 : : {
826 [ - + ]: 5 : if (iculocaleEl)
827 : 0 : errorConflictingDefElem(defel, pstate);
828 : 5 : iculocaleEl = defel;
829 : : }
830 [ + + ]: 520 : else if (strcmp(defel->defname, "icu_rules") == 0)
831 : : {
832 [ - + ]: 1 : if (icurulesEl)
833 : 0 : errorConflictingDefElem(defel, pstate);
834 : 1 : icurulesEl = defel;
835 : : }
836 [ + + ]: 519 : else if (strcmp(defel->defname, "locale_provider") == 0)
837 : : {
838 [ - + ]: 62 : if (locproviderEl)
839 : 0 : errorConflictingDefElem(defel, pstate);
840 : 62 : locproviderEl = defel;
841 : : }
842 [ + + ]: 457 : else if (strcmp(defel->defname, "is_template") == 0)
843 : : {
844 [ - + ]: 57 : if (istemplateEl)
845 : 0 : errorConflictingDefElem(defel, pstate);
846 : 57 : istemplateEl = defel;
847 : : }
848 [ + + ]: 400 : else if (strcmp(defel->defname, "allow_connections") == 0)
849 : : {
850 [ - + ]: 56 : if (allowconnectionsEl)
851 : 0 : errorConflictingDefElem(defel, pstate);
852 : 56 : allowconnectionsEl = defel;
853 : : }
854 [ - + ]: 344 : else if (strcmp(defel->defname, "connection_limit") == 0)
855 : : {
856 [ # # ]: 0 : if (connlimitEl)
857 : 0 : errorConflictingDefElem(defel, pstate);
858 : 0 : connlimitEl = defel;
859 : : }
860 [ + + ]: 344 : else if (strcmp(defel->defname, "collation_version") == 0)
861 : : {
862 [ - + ]: 32 : if (collversionEl)
863 : 0 : errorConflictingDefElem(defel, pstate);
864 : 32 : collversionEl = defel;
865 : : }
866 [ - + ]: 312 : else if (strcmp(defel->defname, "location") == 0)
867 : : {
868 [ # # ]: 0 : ereport(WARNING,
869 : : (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
870 : : errmsg("LOCATION is not supported anymore"),
871 : : errhint("Consider using tablespaces instead."),
872 : : parser_errposition(pstate, defel->location)));
873 : : }
874 [ + + ]: 312 : else if (strcmp(defel->defname, "oid") == 0)
875 : : {
876 : 147 : dboid = defGetObjectId(defel);
877 : :
878 : : /*
879 : : * We don't normally permit new databases to be created with
880 : : * system-assigned OIDs. pg_upgrade tries to preserve database
881 : : * OIDs, so we can't allow any database to be created with an OID
882 : : * that might be in use in a freshly-initialized cluster created
883 : : * by some future version. We assume all such OIDs will be from
884 : : * the system-managed OID range.
885 : : *
886 : : * As an exception, however, we permit any OID to be assigned when
887 : : * allow_system_table_mods=on (so that initdb can assign system
888 : : * OIDs to template0 and postgres) or when performing a binary
889 : : * upgrade (so that pg_upgrade can preserve whatever OIDs it finds
890 : : * in the source cluster).
891 : : */
892 [ + + ]: 147 : if (dboid < FirstNormalObjectId &&
893 [ + + - + ]: 130 : !allowSystemTableMods && !IsBinaryUpgrade)
894 [ # # ]: 0 : ereport(ERROR,
895 : : (errcode(ERRCODE_INVALID_PARAMETER_VALUE)),
896 : : errmsg("OIDs less than %u are reserved for system objects", FirstNormalObjectId));
897 : : }
898 [ + - ]: 165 : else if (strcmp(defel->defname, "strategy") == 0)
899 : : {
900 [ - + ]: 165 : if (strategyEl)
901 : 0 : errorConflictingDefElem(defel, pstate);
902 : 165 : strategyEl = defel;
903 : : }
904 : : else
905 [ # # ]: 0 : ereport(ERROR,
906 : : (errcode(ERRCODE_SYNTAX_ERROR),
907 : : errmsg("option \"%s\" not recognized", defel->defname),
908 : : parser_errposition(pstate, defel->location)));
909 : : }
910 : :
911 [ - + - - ]: 445 : if (ownerEl && ownerEl->arg)
912 : 0 : dbowner = defGetString(ownerEl);
913 [ + + + - ]: 445 : if (templateEl && templateEl->arg)
914 : 197 : dbtemplate = defGetString(templateEl);
915 [ + + + - ]: 445 : if (encodingEl && encodingEl->arg)
916 : : {
917 : : const char *encoding_name;
918 : :
919 [ - + ]: 58 : if (IsA(encodingEl->arg, Integer))
920 : : {
921 : 0 : encoding = defGetInt32(encodingEl);
922 : 0 : encoding_name = pg_encoding_to_char(encoding);
923 [ # # ]: 0 : if (strcmp(encoding_name, "") == 0 ||
924 [ # # ]: 0 : pg_valid_server_encoding(encoding_name) < 0)
925 [ # # ]: 0 : ereport(ERROR,
926 : : (errcode(ERRCODE_UNDEFINED_OBJECT),
927 : : errmsg("%d is not a valid encoding code",
928 : : encoding),
929 : : parser_errposition(pstate, encodingEl->location)));
930 : : }
931 : : else
932 : : {
933 : 58 : encoding_name = defGetString(encodingEl);
934 : 58 : encoding = pg_valid_server_encoding(encoding_name);
935 [ - + ]: 58 : if (encoding < 0)
936 [ # # ]: 0 : ereport(ERROR,
937 : : (errcode(ERRCODE_UNDEFINED_OBJECT),
938 : : errmsg("%s is not a valid encoding name",
939 : : encoding_name),
940 : : parser_errposition(pstate, encodingEl->location)));
941 : : }
942 : : }
943 [ + + + - ]: 445 : if (localeEl && localeEl->arg)
944 : : {
945 : 57 : dbcollate = defGetString(localeEl);
946 : 57 : dbctype = defGetString(localeEl);
947 : 57 : dblocale = defGetString(localeEl);
948 : : }
949 [ + + + - ]: 445 : if (builtinlocaleEl && builtinlocaleEl->arg)
950 : 8 : dblocale = defGetString(builtinlocaleEl);
951 [ + + + - ]: 445 : if (collateEl && collateEl->arg)
952 : 11 : dbcollate = defGetString(collateEl);
953 [ + + + - ]: 445 : if (ctypeEl && ctypeEl->arg)
954 : 11 : dbctype = defGetString(ctypeEl);
955 [ + + + - ]: 445 : if (iculocaleEl && iculocaleEl->arg)
956 : 5 : dblocale = defGetString(iculocaleEl);
957 [ + + + - ]: 445 : if (icurulesEl && icurulesEl->arg)
958 : 1 : dbicurules = defGetString(icurulesEl);
959 [ + + + - ]: 445 : if (locproviderEl && locproviderEl->arg)
960 : : {
961 : 62 : char *locproviderstr = defGetString(locproviderEl);
962 : :
963 [ + + ]: 62 : if (pg_strcasecmp(locproviderstr, "builtin") == 0)
964 : 17 : dblocprovider = COLLPROVIDER_BUILTIN;
965 [ + + ]: 45 : else if (pg_strcasecmp(locproviderstr, "icu") == 0)
966 : 8 : dblocprovider = COLLPROVIDER_ICU;
967 [ + + ]: 37 : else if (pg_strcasecmp(locproviderstr, "libc") == 0)
968 : 36 : dblocprovider = COLLPROVIDER_LIBC;
969 : : else
970 [ + - ]: 1 : ereport(ERROR,
971 : : (errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
972 : : errmsg("unrecognized locale provider: %s",
973 : : locproviderstr)));
974 : : }
975 [ + + + - ]: 444 : if (istemplateEl && istemplateEl->arg)
976 : 57 : dbistemplate = defGetBoolean(istemplateEl);
977 [ + + + - ]: 444 : if (allowconnectionsEl && allowconnectionsEl->arg)
978 : 56 : dballowconnections = defGetBoolean(allowconnectionsEl);
979 [ - + - - ]: 444 : if (connlimitEl && connlimitEl->arg)
980 : : {
981 : 0 : dbconnlimit = defGetInt32(connlimitEl);
982 [ # # ]: 0 : if (dbconnlimit < DATCONNLIMIT_UNLIMITED)
983 [ # # ]: 0 : ereport(ERROR,
984 : : (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
985 : : errmsg("invalid connection limit: %d", dbconnlimit)));
986 : : }
987 [ + + ]: 444 : if (collversionEl)
988 : 32 : dbcollversion = defGetString(collversionEl);
989 : :
990 : : /* obtain OID of proposed owner */
991 [ - + ]: 444 : if (dbowner)
992 : 0 : datdba = get_role_oid(dbowner, false);
993 : : else
994 : 444 : datdba = GetUserId();
995 : :
996 : : /*
997 : : * To create a database, must have createdb privilege and must be able to
998 : : * become the target role (this does not imply that the target role itself
999 : : * must have createdb privilege). The latter provision guards against
1000 : : * "giveaway" attacks. Note that a superuser will always have both of
1001 : : * these privileges a fortiori.
1002 : : */
1003 [ + + ]: 444 : if (!have_createdb_privilege())
1004 [ + - ]: 4 : ereport(ERROR,
1005 : : (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
1006 : : errmsg("permission denied to create database")));
1007 : :
1008 : 440 : check_can_set_role(GetUserId(), datdba);
1009 : :
1010 : : /*
1011 : : * Lookup database (template) to be cloned, and obtain share lock on it.
1012 : : * ShareLock allows two CREATE DATABASEs to work from the same template
1013 : : * concurrently, while ensuring no one is busy dropping it in parallel
1014 : : * (which would be Very Bad since we'd likely get an incomplete copy
1015 : : * without knowing it). This also prevents any new connections from being
1016 : : * made to the source until we finish copying it, so we can be sure it
1017 : : * won't change underneath us.
1018 : : */
1019 [ + + ]: 440 : if (!dbtemplate)
1020 : 244 : dbtemplate = "template1"; /* Default template database name */
1021 : :
1022 [ - + ]: 440 : if (!get_db_info(dbtemplate, ShareLock,
1023 : : &src_dboid, &src_owner, &src_encoding,
1024 : : &src_istemplate, &src_allowconn, &src_hasloginevt,
1025 : : &src_frozenxid, &src_minmxid, &src_deftablespace,
1026 : : &src_collate, &src_ctype, &src_locale, &src_icurules, &src_locprovider,
1027 : : &src_collversion))
1028 [ # # ]: 0 : ereport(ERROR,
1029 : : (errcode(ERRCODE_UNDEFINED_DATABASE),
1030 : : errmsg("template database \"%s\" does not exist",
1031 : : dbtemplate)));
1032 : :
1033 : : /*
1034 : : * If the source database was in the process of being dropped, we can't
1035 : : * use it as a template.
1036 : : */
1037 [ + + ]: 440 : if (database_is_invalid_oid(src_dboid))
1038 [ + - ]: 1 : ereport(ERROR,
1039 : : errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1040 : : errmsg("cannot use invalid database \"%s\" as template", dbtemplate),
1041 : : errhint("Use DROP DATABASE to drop invalid databases."));
1042 : :
1043 : : /*
1044 : : * Permission check: to copy a DB that's not marked datistemplate, you
1045 : : * must be superuser or the owner thereof.
1046 : : */
1047 [ + + ]: 439 : if (!src_istemplate)
1048 : : {
1049 [ - + ]: 13 : if (!object_ownercheck(DatabaseRelationId, src_dboid, GetUserId()))
1050 [ # # ]: 0 : ereport(ERROR,
1051 : : (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
1052 : : errmsg("permission denied to copy database \"%s\"",
1053 : : dbtemplate)));
1054 : : }
1055 : :
1056 : : /* Validate the database creation strategy. */
1057 [ + + + - ]: 439 : if (strategyEl && strategyEl->arg)
1058 : : {
1059 : : char *strategy;
1060 : :
1061 : 165 : strategy = defGetString(strategyEl);
1062 [ + + ]: 165 : if (pg_strcasecmp(strategy, "wal_log") == 0)
1063 : 11 : dbstrategy = CREATEDB_WAL_LOG;
1064 [ + + ]: 154 : else if (pg_strcasecmp(strategy, "file_copy") == 0)
1065 : : {
1066 : : /*
1067 : : * If data checksums are being enabled we must not use file_copy
1068 : : * since it might copy source database which hasn't yet had data
1069 : : * checksums enabled, and the destination database will be skipped
1070 : : * as it's expected to have data checksums enabled. Once we have
1071 : : * an XID assigned this needs to be rechecked, but if can error
1072 : : * out already we can save a lot of work.
1073 : : */
1074 [ - + ]: 153 : if (DataChecksumsInProgressOn())
1075 [ # # ]: 0 : ereport(ERROR,
1076 : : errcode(ERRCODE_INVALID_PARAMETER_VALUE),
1077 : : errmsg("create database strategy \"%s\" not allowed when data checksums are being enabled",
1078 : : strategy));
1079 : 153 : dbstrategy = CREATEDB_FILE_COPY;
1080 : : }
1081 : : else
1082 [ + - ]: 1 : ereport(ERROR,
1083 : : (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
1084 : : errmsg("invalid create database strategy \"%s\"", strategy),
1085 : : errhint("Valid strategies are \"wal_log\" and \"file_copy\".")));
1086 : : }
1087 : :
1088 : : /* If encoding or locales are defaulted, use source's setting */
1089 [ + + ]: 438 : if (encoding < 0)
1090 : 380 : encoding = src_encoding;
1091 [ + + ]: 438 : if (dbcollate == NULL)
1092 : 373 : dbcollate = src_collate;
1093 [ + + ]: 438 : if (dbctype == NULL)
1094 : 373 : dbctype = src_ctype;
1095 [ + + ]: 438 : if (dblocprovider == '\0')
1096 : 377 : dblocprovider = src_locprovider;
1097 [ + + + + ]: 438 : if (dblocale == NULL && dblocprovider == src_locprovider)
1098 : 373 : dblocale = src_locale;
1099 [ + + ]: 438 : if (dbicurules == NULL)
1100 : 437 : dbicurules = src_icurules;
1101 : :
1102 : : /* Some encodings are client only */
1103 [ + - + - : 438 : if (!PG_VALID_BE_ENCODING(encoding))
- + ]
1104 [ # # ]: 0 : ereport(ERROR,
1105 : : (errcode(ERRCODE_WRONG_OBJECT_TYPE),
1106 : : errmsg("invalid server encoding %d", encoding)));
1107 : :
1108 : : /* Check that the chosen locales are valid, and get canonical spellings */
1109 [ + + ]: 438 : if (!check_locale(LC_COLLATE, dbcollate, &canonname))
1110 : : {
1111 [ - + ]: 1 : if (dblocprovider == COLLPROVIDER_BUILTIN)
1112 [ # # ]: 0 : ereport(ERROR,
1113 : : (errcode(ERRCODE_WRONG_OBJECT_TYPE),
1114 : : errmsg("invalid LC_COLLATE locale name: \"%s\"", dbcollate),
1115 : : errhint("If the locale name is specific to the builtin provider, use BUILTIN_LOCALE.")));
1116 [ - + ]: 1 : else if (dblocprovider == COLLPROVIDER_ICU)
1117 [ # # ]: 0 : ereport(ERROR,
1118 : : (errcode(ERRCODE_WRONG_OBJECT_TYPE),
1119 : : errmsg("invalid LC_COLLATE locale name: \"%s\"", dbcollate),
1120 : : errhint("If the locale name is specific to the ICU provider, use ICU_LOCALE.")));
1121 : : else
1122 [ + - ]: 1 : ereport(ERROR,
1123 : : (errcode(ERRCODE_WRONG_OBJECT_TYPE),
1124 : : errmsg("invalid LC_COLLATE locale name: \"%s\"", dbcollate)));
1125 : : }
1126 : 437 : dbcollate = canonname;
1127 [ + + ]: 437 : if (!check_locale(LC_CTYPE, dbctype, &canonname))
1128 : : {
1129 [ - + ]: 1 : if (dblocprovider == COLLPROVIDER_BUILTIN)
1130 [ # # ]: 0 : ereport(ERROR,
1131 : : (errcode(ERRCODE_WRONG_OBJECT_TYPE),
1132 : : errmsg("invalid LC_CTYPE locale name: \"%s\"", dbctype),
1133 : : errhint("If the locale name is specific to the builtin provider, use BUILTIN_LOCALE.")));
1134 [ - + ]: 1 : else if (dblocprovider == COLLPROVIDER_ICU)
1135 [ # # ]: 0 : ereport(ERROR,
1136 : : (errcode(ERRCODE_WRONG_OBJECT_TYPE),
1137 : : errmsg("invalid LC_CTYPE locale name: \"%s\"", dbctype),
1138 : : errhint("If the locale name is specific to the ICU provider, use ICU_LOCALE.")));
1139 : : else
1140 [ + - ]: 1 : ereport(ERROR,
1141 : : (errcode(ERRCODE_WRONG_OBJECT_TYPE),
1142 : : errmsg("invalid LC_CTYPE locale name: \"%s\"", dbctype)));
1143 : : }
1144 : :
1145 : 436 : dbctype = canonname;
1146 : :
1147 : 436 : check_encoding_locale_matches(encoding, dbcollate, dbctype);
1148 : :
1149 : : /* validate provider-specific parameters */
1150 [ + + ]: 436 : if (dblocprovider != COLLPROVIDER_BUILTIN)
1151 : : {
1152 [ - + ]: 405 : if (builtinlocaleEl)
1153 [ # # ]: 0 : ereport(ERROR,
1154 : : (errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
1155 : : errmsg("BUILTIN_LOCALE cannot be specified unless locale provider is builtin")));
1156 : : }
1157 : :
1158 [ + + ]: 436 : if (dblocprovider != COLLPROVIDER_ICU)
1159 : : {
1160 [ + + ]: 421 : if (iculocaleEl)
1161 [ + - ]: 1 : ereport(ERROR,
1162 : : (errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
1163 : : errmsg("ICU locale cannot be specified unless locale provider is ICU")));
1164 : :
1165 [ + + ]: 420 : if (dbicurules)
1166 [ + - ]: 1 : ereport(ERROR,
1167 : : (errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
1168 : : errmsg("ICU rules cannot be specified unless locale provider is ICU")));
1169 : : }
1170 : :
1171 : : /* validate and canonicalize locale for the provider */
1172 [ + + ]: 434 : if (dblocprovider == COLLPROVIDER_BUILTIN)
1173 : : {
1174 : : /*
1175 : : * This would happen if template0 uses the libc provider but the new
1176 : : * database uses builtin.
1177 : : */
1178 [ + + ]: 29 : if (!dblocale)
1179 [ + - ]: 1 : ereport(ERROR,
1180 : : (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
1181 : : errmsg("LOCALE or BUILTIN_LOCALE must be specified")));
1182 : :
1183 : 28 : dblocale = builtin_validate_locale(encoding, dblocale);
1184 : : }
1185 [ + + ]: 405 : else if (dblocprovider == COLLPROVIDER_ICU)
1186 : : {
1187 [ + + ]: 15 : if (!(is_encoding_supported_by_icu(encoding)))
1188 [ + - ]: 1 : ereport(ERROR,
1189 : : (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
1190 : : errmsg("encoding \"%s\" is not supported with ICU provider",
1191 : : pg_encoding_to_char(encoding))));
1192 : :
1193 : : /*
1194 : : * This would happen if template0 uses the libc provider but the new
1195 : : * database uses icu.
1196 : : */
1197 [ + + ]: 14 : if (!dblocale)
1198 [ + - ]: 1 : ereport(ERROR,
1199 : : (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
1200 : : errmsg("LOCALE or ICU_LOCALE must be specified")));
1201 : :
1202 : : /*
1203 : : * During binary upgrade, or when the locale came from the template
1204 : : * database, preserve locale string. Otherwise, canonicalize to a
1205 : : * language tag.
1206 : : */
1207 [ + - + + ]: 13 : if (!IsBinaryUpgrade && dblocale != src_locale)
1208 : : {
1209 : 7 : char *langtag = icu_language_tag(dblocale,
1210 : : icu_validation_level);
1211 : :
1212 [ + + + + ]: 7 : if (langtag && strcmp(dblocale, langtag) != 0)
1213 : : {
1214 [ + - ]: 3 : ereport(NOTICE,
1215 : : (errmsg("using standard form \"%s\" for ICU locale \"%s\"",
1216 : : langtag, dblocale)));
1217 : :
1218 : 3 : dblocale = langtag;
1219 : : }
1220 : : }
1221 : :
1222 : 13 : icu_validate_locale(dblocale);
1223 : : }
1224 : :
1225 : : /* for libc, locale comes from datcollate and datctype */
1226 [ + + ]: 429 : if (dblocprovider == COLLPROVIDER_LIBC)
1227 : 390 : dblocale = NULL;
1228 : :
1229 : : /*
1230 : : * Check that the new encoding and locale settings match the source
1231 : : * database. We insist on this because we simply copy the source data ---
1232 : : * any non-ASCII data would be wrongly encoded, and any indexes sorted
1233 : : * according to the source locale would be wrong.
1234 : : *
1235 : : * However, we assume that template0 doesn't contain any non-ASCII data
1236 : : * nor any indexes that depend on collation or ctype, so template0 can be
1237 : : * used as template for creating a database with any encoding or locale.
1238 : : */
1239 [ + + ]: 429 : if (strcmp(dbtemplate, "template0") != 0)
1240 : : {
1241 [ - + ]: 258 : if (encoding != src_encoding)
1242 [ # # ]: 0 : ereport(ERROR,
1243 : : (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
1244 : : errmsg("new encoding (%s) is incompatible with the encoding of the template database (%s)",
1245 : : pg_encoding_to_char(encoding),
1246 : : pg_encoding_to_char(src_encoding)),
1247 : : errhint("Use the same encoding as in the template database, or use template0 as template.")));
1248 : :
1249 [ + + ]: 258 : if (strcmp(dbcollate, src_collate) != 0)
1250 [ + - ]: 1 : ereport(ERROR,
1251 : : (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
1252 : : errmsg("new collation (%s) is incompatible with the collation of the template database (%s)",
1253 : : dbcollate, src_collate),
1254 : : errhint("Use the same collation as in the template database, or use template0 as template.")));
1255 : :
1256 [ - + ]: 257 : if (strcmp(dbctype, src_ctype) != 0)
1257 [ # # ]: 0 : ereport(ERROR,
1258 : : (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
1259 : : errmsg("new LC_CTYPE (%s) is incompatible with the LC_CTYPE of the template database (%s)",
1260 : : dbctype, src_ctype),
1261 : : errhint("Use the same LC_CTYPE as in the template database, or use template0 as template.")));
1262 : :
1263 [ - + ]: 257 : if (dblocprovider != src_locprovider)
1264 [ # # ]: 0 : ereport(ERROR,
1265 : : (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
1266 : : errmsg("new locale provider (%s) does not match locale provider of the template database (%s)",
1267 : : collprovider_name(dblocprovider), collprovider_name(src_locprovider)),
1268 : : errhint("Use the same locale provider as in the template database, or use template0 as template.")));
1269 : :
1270 [ + + ]: 257 : if (dblocprovider == COLLPROVIDER_ICU)
1271 : : {
1272 : : char *val1;
1273 : : char *val2;
1274 : :
1275 : : Assert(dblocale);
1276 : : Assert(src_locale);
1277 [ - + ]: 6 : if (strcmp(dblocale, src_locale) != 0)
1278 [ # # ]: 0 : ereport(ERROR,
1279 : : (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
1280 : : errmsg("new ICU locale (%s) is incompatible with the ICU locale of the template database (%s)",
1281 : : dblocale, src_locale),
1282 : : errhint("Use the same ICU locale as in the template database, or use template0 as template.")));
1283 : :
1284 : 6 : val1 = dbicurules;
1285 [ + - ]: 6 : if (!val1)
1286 : 6 : val1 = "";
1287 : 6 : val2 = src_icurules;
1288 [ + - ]: 6 : if (!val2)
1289 : 6 : val2 = "";
1290 [ - + ]: 6 : if (strcmp(val1, val2) != 0)
1291 [ # # ]: 0 : ereport(ERROR,
1292 : : (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
1293 : : errmsg("new ICU collation rules (%s) are incompatible with the ICU collation rules of the template database (%s)",
1294 : : val1, val2),
1295 : : errhint("Use the same ICU collation rules as in the template database, or use template0 as template.")));
1296 : : }
1297 : : }
1298 : :
1299 : : /*
1300 : : * If we got a collation version for the template database, check that it
1301 : : * matches the actual OS collation version. Otherwise error; the user
1302 : : * needs to fix the template database first. Don't complain if a
1303 : : * collation version was specified explicitly as a statement option; that
1304 : : * is used by pg_upgrade to reproduce the old state exactly.
1305 : : *
1306 : : * (If the template database has no collation version, then either the
1307 : : * platform/provider does not support collation versioning, or it's
1308 : : * template0, for which we stipulate that it does not contain
1309 : : * collation-using objects.)
1310 : : */
1311 [ + + + - ]: 428 : if (src_collversion && !collversionEl)
1312 : : {
1313 : : char *actual_versionstr;
1314 : : const char *locale;
1315 : :
1316 [ + + ]: 174 : if (dblocprovider == COLLPROVIDER_LIBC)
1317 : 163 : locale = dbcollate;
1318 : : else
1319 : 11 : locale = dblocale;
1320 : :
1321 : 174 : actual_versionstr = get_collation_actual_version(dblocprovider, locale);
1322 [ - + ]: 174 : if (!actual_versionstr)
1323 [ # # ]: 0 : ereport(ERROR,
1324 : : (errmsg("template database \"%s\" has a collation version, but no actual collation version could be determined",
1325 : : dbtemplate)));
1326 : :
1327 [ - + ]: 174 : if (strcmp(actual_versionstr, src_collversion) != 0)
1328 [ # # ]: 0 : ereport(ERROR,
1329 : : (errmsg("template database \"%s\" has a collation version mismatch",
1330 : : dbtemplate),
1331 : : errdetail("The template database was created using collation version %s, "
1332 : : "but the operating system provides version %s.",
1333 : : src_collversion, actual_versionstr),
1334 : : errhint("Rebuild all objects in the template database that use the default collation and run "
1335 : : "ALTER DATABASE %s REFRESH COLLATION VERSION, "
1336 : : "or build PostgreSQL with the right library version.",
1337 : : quote_identifier(dbtemplate))));
1338 : : }
1339 : :
1340 [ + + ]: 428 : if (dbcollversion == NULL)
1341 : 396 : dbcollversion = src_collversion;
1342 : :
1343 : : /*
1344 : : * Normally, we copy the collation version from the template database.
1345 : : * This last resort only applies if the template database does not have a
1346 : : * collation version, which is normally only the case for template0.
1347 : : */
1348 [ + + ]: 428 : if (dbcollversion == NULL)
1349 : : {
1350 : : const char *locale;
1351 : :
1352 [ + + ]: 222 : if (dblocprovider == COLLPROVIDER_LIBC)
1353 : 201 : locale = dbcollate;
1354 : : else
1355 : 21 : locale = dblocale;
1356 : :
1357 : 222 : dbcollversion = get_collation_actual_version(dblocprovider, locale);
1358 : : }
1359 : :
1360 : : /* Resolve default tablespace for new database */
1361 [ + + + - ]: 428 : if (tablespacenameEl && tablespacenameEl->arg)
1362 : 17 : {
1363 : : char *tablespacename;
1364 : : AclResult aclresult;
1365 : :
1366 : 17 : tablespacename = defGetString(tablespacenameEl);
1367 : 17 : dst_deftablespace = get_tablespace_oid(tablespacename, false);
1368 : : /* check permissions */
1369 : 17 : aclresult = object_aclcheck(TableSpaceRelationId, dst_deftablespace, GetUserId(),
1370 : : ACL_CREATE);
1371 [ - + ]: 17 : if (aclresult != ACLCHECK_OK)
1372 : 0 : aclcheck_error(aclresult, OBJECT_TABLESPACE,
1373 : : tablespacename);
1374 : :
1375 : : /* pg_global must never be the default tablespace */
1376 [ - + ]: 17 : if (dst_deftablespace == GLOBALTABLESPACE_OID)
1377 [ # # ]: 0 : ereport(ERROR,
1378 : : (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
1379 : : errmsg("pg_global cannot be used as default tablespace")));
1380 : :
1381 : : /*
1382 : : * If we are trying to change the default tablespace of the template,
1383 : : * we require that the template not have any files in the new default
1384 : : * tablespace. This is necessary because otherwise the copied
1385 : : * database would contain pg_class rows that refer to its default
1386 : : * tablespace both explicitly (by OID) and implicitly (as zero), which
1387 : : * would cause problems. For example another CREATE DATABASE using
1388 : : * the copied database as template, and trying to change its default
1389 : : * tablespace again, would yield outright incorrect results (it would
1390 : : * improperly move tables to the new default tablespace that should
1391 : : * stay in the same tablespace).
1392 : : */
1393 [ + - ]: 17 : if (dst_deftablespace != src_deftablespace)
1394 : : {
1395 : : char *srcpath;
1396 : : struct stat st;
1397 : :
1398 : 17 : srcpath = GetDatabasePath(src_dboid, dst_deftablespace);
1399 : :
1400 [ - + ]: 17 : if (stat(srcpath, &st) == 0 &&
1401 [ # # ]: 0 : S_ISDIR(st.st_mode) &&
1402 [ # # ]: 0 : !directory_is_empty(srcpath))
1403 [ # # ]: 0 : ereport(ERROR,
1404 : : (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
1405 : : errmsg("cannot assign new default tablespace \"%s\"",
1406 : : tablespacename),
1407 : : errdetail("There is a conflict because database \"%s\" already has some tables in this tablespace.",
1408 : : dbtemplate)));
1409 : 17 : pfree(srcpath);
1410 : : }
1411 : : }
1412 : : else
1413 : : {
1414 : : /* Use template database's default tablespace */
1415 : 411 : dst_deftablespace = src_deftablespace;
1416 : : /* Note there is no additional permission check in this path */
1417 : : }
1418 : :
1419 : : /*
1420 : : * If built with appropriate switch, whine when regression-testing
1421 : : * conventions for database names are violated. But don't complain during
1422 : : * initdb.
1423 : : */
1424 : : #ifdef ENFORCE_REGRESSION_TEST_NAME_RESTRICTIONS
1425 : : if (IsUnderPostmaster && strstr(dbname, "regression") == NULL)
1426 : : elog(WARNING, "databases created by regression test cases should have names including \"regression\"");
1427 : : #endif
1428 : :
1429 : : /*
1430 : : * Check for db name conflict. This is just to give a more friendly error
1431 : : * message than "unique index violation". There's a race condition but
1432 : : * we're willing to accept the less friendly message in that case.
1433 : : */
1434 [ + + ]: 428 : if (OidIsValid(get_database_oid(dbname, true)))
1435 [ + - ]: 1 : ereport(ERROR,
1436 : : (errcode(ERRCODE_DUPLICATE_DATABASE),
1437 : : errmsg("database \"%s\" already exists", dbname)));
1438 : :
1439 : : /*
1440 : : * The source DB can't have any active backends, except this one
1441 : : * (exception is to allow CREATE DB while connected to template1).
1442 : : * Otherwise we might copy inconsistent data.
1443 : : *
1444 : : * This should be last among the basic error checks, because it involves
1445 : : * potential waiting; we may as well throw an error first if we're gonna
1446 : : * throw one.
1447 : : */
1448 [ + + ]: 427 : if (CountOtherDBBackends(src_dboid, ¬herbackends, &npreparedxacts))
1449 [ + - ]: 1 : ereport(ERROR,
1450 : : (errcode(ERRCODE_OBJECT_IN_USE),
1451 : : errmsg("source database \"%s\" is being accessed by other users",
1452 : : dbtemplate),
1453 : : errdetail_busy_db(notherbackends, npreparedxacts)));
1454 : :
1455 : : /*
1456 : : * Select an OID for the new database, checking that it doesn't have a
1457 : : * filename conflict with anything already existing in the tablespace
1458 : : * directories.
1459 : : */
1460 : 426 : pg_database_rel = table_open(DatabaseRelationId, RowExclusiveLock);
1461 : :
1462 : : /*
1463 : : * If database OID is configured, check if the OID is already in use or
1464 : : * data directory already exists.
1465 : : */
1466 [ + + ]: 426 : if (OidIsValid(dboid))
1467 : : {
1468 : 147 : char *existing_dbname = get_database_name(dboid);
1469 : :
1470 [ - + ]: 147 : if (existing_dbname != NULL)
1471 [ # # ]: 0 : ereport(ERROR,
1472 : : (errcode(ERRCODE_INVALID_PARAMETER_VALUE)),
1473 : : errmsg("database OID %u is already in use by database \"%s\"",
1474 : : dboid, existing_dbname));
1475 : :
1476 [ - + ]: 147 : if (check_db_file_conflict(dboid))
1477 [ # # ]: 0 : ereport(ERROR,
1478 : : (errcode(ERRCODE_INVALID_PARAMETER_VALUE)),
1479 : : errmsg("data directory with the specified OID %u already exists", dboid));
1480 : : }
1481 : : else
1482 : : {
1483 : : /* Select an OID for the new database if is not explicitly configured. */
1484 : : do
1485 : : {
1486 : 279 : dboid = GetNewOidWithIndex(pg_database_rel, DatabaseOidIndexId,
1487 : : Anum_pg_database_oid);
1488 [ - + ]: 279 : } while (check_db_file_conflict(dboid));
1489 : : }
1490 : :
1491 : : /*
1492 : : * Insert a new tuple into pg_database. This establishes our ownership of
1493 : : * the new database name (anyone else trying to insert the same name will
1494 : : * block on the unique index, and fail after we commit).
1495 : : */
1496 : :
1497 : : Assert((dblocprovider != COLLPROVIDER_LIBC && dblocale) ||
1498 : : (dblocprovider == COLLPROVIDER_LIBC && !dblocale));
1499 : :
1500 : : /* Form tuple */
1501 : 426 : new_record[Anum_pg_database_oid - 1] = ObjectIdGetDatum(dboid);
1502 : 426 : new_record[Anum_pg_database_datname - 1] =
1503 : 426 : DirectFunctionCall1(namein, CStringGetDatum(dbname));
1504 : 426 : new_record[Anum_pg_database_datdba - 1] = ObjectIdGetDatum(datdba);
1505 : 426 : new_record[Anum_pg_database_encoding - 1] = Int32GetDatum(encoding);
1506 : 426 : new_record[Anum_pg_database_datlocprovider - 1] = CharGetDatum(dblocprovider);
1507 : 426 : new_record[Anum_pg_database_datistemplate - 1] = BoolGetDatum(dbistemplate);
1508 : 426 : new_record[Anum_pg_database_datallowconn - 1] = BoolGetDatum(dballowconnections);
1509 : 426 : new_record[Anum_pg_database_dathasloginevt - 1] = BoolGetDatum(src_hasloginevt);
1510 : 426 : new_record[Anum_pg_database_datconnlimit - 1] = Int32GetDatum(dbconnlimit);
1511 : 426 : new_record[Anum_pg_database_datfrozenxid - 1] = TransactionIdGetDatum(src_frozenxid);
1512 : 426 : new_record[Anum_pg_database_datminmxid - 1] = TransactionIdGetDatum(src_minmxid);
1513 : 426 : new_record[Anum_pg_database_dattablespace - 1] = ObjectIdGetDatum(dst_deftablespace);
1514 : 426 : new_record[Anum_pg_database_datcollate - 1] = CStringGetTextDatum(dbcollate);
1515 : 426 : new_record[Anum_pg_database_datctype - 1] = CStringGetTextDatum(dbctype);
1516 [ + + ]: 426 : if (dblocale)
1517 : 38 : new_record[Anum_pg_database_datlocale - 1] = CStringGetTextDatum(dblocale);
1518 : : else
1519 : 388 : new_record_nulls[Anum_pg_database_datlocale - 1] = true;
1520 [ - + ]: 426 : if (dbicurules)
1521 : 0 : new_record[Anum_pg_database_daticurules - 1] = CStringGetTextDatum(dbicurules);
1522 : : else
1523 : 426 : new_record_nulls[Anum_pg_database_daticurules - 1] = true;
1524 [ + + ]: 426 : if (dbcollversion)
1525 : 369 : new_record[Anum_pg_database_datcollversion - 1] = CStringGetTextDatum(dbcollversion);
1526 : : else
1527 : 57 : new_record_nulls[Anum_pg_database_datcollversion - 1] = true;
1528 : :
1529 : : /*
1530 : : * We deliberately set datacl to default (NULL), rather than copying it
1531 : : * from the template database. Copying it would be a bad idea when the
1532 : : * owner is not the same as the template's owner.
1533 : : */
1534 : 426 : new_record_nulls[Anum_pg_database_datacl - 1] = true;
1535 : :
1536 : 426 : tuple = heap_form_tuple(RelationGetDescr(pg_database_rel),
1537 : : new_record, new_record_nulls);
1538 : :
1539 : 426 : INJECTION_POINT("createdb-before-catalog-insert", NULL);
1540 : :
1541 : 426 : CatalogTupleInsert(pg_database_rel, tuple);
1542 : :
1543 : : /*
1544 : : * Now generate additional catalog entries associated with the new DB
1545 : : */
1546 : :
1547 : : /* Register owner dependency */
1548 : 426 : recordDependencyOnOwner(DatabaseRelationId, dboid, datdba);
1549 : :
1550 : : /* Create pg_shdepend entries for objects within database */
1551 : 426 : copyTemplateDependencies(src_dboid, dboid);
1552 : :
1553 : : /* Post creation hook for new database */
1554 [ - + ]: 426 : InvokeObjectPostCreateHook(DatabaseRelationId, dboid, 0);
1555 : :
1556 : : /*
1557 : : * If we're going to be reading data for the to-be-created database into
1558 : : * shared_buffers, take a lock on it. Nobody should know that this
1559 : : * database exists yet, but it's good to maintain the invariant that an
1560 : : * AccessExclusiveLock on the database is sufficient to drop all of its
1561 : : * buffers without worrying about more being read later.
1562 : : *
1563 : : * Note that we need to do this before entering the
1564 : : * PG_ENSURE_ERROR_CLEANUP block below, because createdb_failure_callback
1565 : : * expects this lock to be held already.
1566 : : */
1567 [ + + ]: 426 : if (dbstrategy == CREATEDB_WAL_LOG)
1568 : 273 : LockSharedObject(DatabaseRelationId, dboid, 0, AccessShareLock);
1569 : :
1570 : : /*
1571 : : * Once we start copying subdirectories, we need to be able to clean 'em
1572 : : * up if we fail. Use an ENSURE block to make sure this happens. (This
1573 : : * is not a 100% solution, because of the possibility of failure during
1574 : : * transaction commit after we leave this routine, but it should handle
1575 : : * most scenarios.)
1576 : : */
1577 : 426 : fparms.src_dboid = src_dboid;
1578 : 426 : fparms.dest_dboid = dboid;
1579 : 426 : fparms.strategy = dbstrategy;
1580 : :
1581 [ + + ]: 426 : PG_ENSURE_ERROR_CLEANUP(createdb_failure_callback,
1582 : : PointerGetDatum(&fparms));
1583 : : {
1584 : : /*
1585 : : * If the user has asked to create a database with WAL_LOG strategy
1586 : : * then call CreateDatabaseUsingWalLog, which will copy the database
1587 : : * at the block level and it will WAL log each copied block.
1588 : : * Otherwise, call CreateDatabaseUsingFileCopy that will copy the
1589 : : * database file by file.
1590 : : */
1591 [ + + ]: 426 : if (dbstrategy == CREATEDB_WAL_LOG)
1592 : 273 : CreateDatabaseUsingWalLog(src_dboid, dboid, src_deftablespace,
1593 : : dst_deftablespace);
1594 : : else
1595 : 153 : CreateDatabaseUsingFileCopy(src_dboid, dboid, src_deftablespace,
1596 : : dst_deftablespace);
1597 : :
1598 : : /*
1599 : : * Close pg_database, but keep lock till commit.
1600 : : */
1601 : 423 : table_close(pg_database_rel, NoLock);
1602 : :
1603 : : /*
1604 : : * Force synchronous commit, thus minimizing the window between
1605 : : * creation of the database files and committal of the transaction. If
1606 : : * we crash before committing, we'll have a DB that's taking up disk
1607 : : * space but is not in pg_database, which is not good.
1608 : : */
1609 : 423 : ForceSyncCommit();
1610 : : }
1611 [ - + ]: 426 : PG_END_ENSURE_ERROR_CLEANUP(createdb_failure_callback,
1612 : : PointerGetDatum(&fparms));
1613 : :
1614 : 423 : return dboid;
1615 : : }
1616 : :
1617 : : /*
1618 : : * Check whether chosen encoding matches chosen locale settings. This
1619 : : * restriction is necessary because libc's locale-specific code usually
1620 : : * fails when presented with data in an encoding it's not expecting. We
1621 : : * allow mismatch in four cases:
1622 : : *
1623 : : * 1. locale encoding = SQL_ASCII, which means that the locale is C/POSIX
1624 : : * which works with any encoding.
1625 : : *
1626 : : * 2. locale encoding = -1, which means that we couldn't determine the
1627 : : * locale's encoding and have to trust the user to get it right.
1628 : : *
1629 : : * 3. selected encoding is UTF8 and platform is win32. This is because
1630 : : * UTF8 is a pseudo codepage that is supported in all locales since it's
1631 : : * converted to UTF16 before being used.
1632 : : *
1633 : : * 4. selected encoding is SQL_ASCII, but only if you're a superuser. This
1634 : : * is risky but we have historically allowed it --- notably, the
1635 : : * regression tests require it.
1636 : : *
1637 : : * Note: if you change this policy, fix initdb to match.
1638 : : */
1639 : : void
1640 : 452 : check_encoding_locale_matches(int encoding, const char *collate, const char *ctype)
1641 : : {
1642 : 452 : int ctype_encoding = pg_get_encoding_from_locale(ctype, true);
1643 : 452 : int collate_encoding = pg_get_encoding_from_locale(collate, true);
1644 : :
1645 [ + + + + : 456 : if (!(ctype_encoding == encoding ||
+ - ]
1646 [ + - ]: 4 : ctype_encoding == PG_SQL_ASCII ||
1647 : : ctype_encoding == -1 ||
1648 : : #ifdef WIN32
1649 : : encoding == PG_UTF8 ||
1650 : : #endif
1651 [ - + ]: 4 : (encoding == PG_SQL_ASCII && superuser())))
1652 [ # # ]: 0 : ereport(ERROR,
1653 : : (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
1654 : : errmsg("encoding \"%s\" does not match locale \"%s\"",
1655 : : pg_encoding_to_char(encoding),
1656 : : ctype),
1657 : : errdetail("The chosen LC_CTYPE setting requires encoding \"%s\".",
1658 : : pg_encoding_to_char(ctype_encoding))));
1659 : :
1660 [ + + + + : 456 : if (!(collate_encoding == encoding ||
+ - ]
1661 [ + - ]: 4 : collate_encoding == PG_SQL_ASCII ||
1662 : : collate_encoding == -1 ||
1663 : : #ifdef WIN32
1664 : : encoding == PG_UTF8 ||
1665 : : #endif
1666 [ - + ]: 4 : (encoding == PG_SQL_ASCII && superuser())))
1667 [ # # ]: 0 : ereport(ERROR,
1668 : : (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
1669 : : errmsg("encoding \"%s\" does not match locale \"%s\"",
1670 : : pg_encoding_to_char(encoding),
1671 : : collate),
1672 : : errdetail("The chosen LC_COLLATE setting requires encoding \"%s\".",
1673 : : pg_encoding_to_char(collate_encoding))));
1674 : 452 : }
1675 : :
1676 : : /* Error cleanup callback for createdb */
1677 : : static void
1678 : 3 : createdb_failure_callback(int code, Datum arg)
1679 : : {
1680 : 3 : createdb_failure_params *fparms = (createdb_failure_params *) DatumGetPointer(arg);
1681 : :
1682 : : /*
1683 : : * If we were copying database at block levels then drop pages for the
1684 : : * destination database that are in the shared buffer cache. And tell
1685 : : * checkpointer to forget any pending fsync and unlink requests for files
1686 : : * in the database. The reasoning behind doing this is same as explained
1687 : : * in dropdb function. But unlike dropdb we don't need to call
1688 : : * pgstat_drop_database because this database is still not created so
1689 : : * there should not be any stat for this.
1690 : : */
1691 [ + + ]: 3 : if (fparms->strategy == CREATEDB_WAL_LOG)
1692 : : {
1693 : 2 : DropDatabaseBuffers(fparms->dest_dboid);
1694 : 2 : ForgetDatabaseSyncRequests(fparms->dest_dboid);
1695 : :
1696 : : /* Release lock on the target database. */
1697 : 2 : UnlockSharedObject(DatabaseRelationId, fparms->dest_dboid, 0,
1698 : : AccessShareLock);
1699 : : }
1700 : :
1701 : : /*
1702 : : * Release lock on source database before doing recursive remove. This is
1703 : : * not essential but it seems desirable to release the lock as soon as
1704 : : * possible.
1705 : : */
1706 : 3 : UnlockSharedObject(DatabaseRelationId, fparms->src_dboid, 0, ShareLock);
1707 : :
1708 : : /* Throw away any successfully copied subdirectories */
1709 : 3 : remove_dbtablespaces(fparms->dest_dboid);
1710 : 3 : }
1711 : :
1712 : :
1713 : : /*
1714 : : * DROP DATABASE
1715 : : */
1716 : : void
1717 : 78 : dropdb(const char *dbname, bool missing_ok, bool force)
1718 : : {
1719 : : Oid db_id;
1720 : : bool db_istemplate;
1721 : : Relation pgdbrel;
1722 : : HeapTuple tup;
1723 : : ScanKeyData scankey;
1724 : : void *inplace_state;
1725 : : Form_pg_database datform;
1726 : : int notherbackends;
1727 : : int npreparedxacts;
1728 : : int nslots,
1729 : : nslots_active;
1730 : : int nsubscriptions;
1731 : :
1732 : : /*
1733 : : * Look up the target database's OID, and get exclusive lock on it. We
1734 : : * need this to ensure that no new backend starts up in the target
1735 : : * database while we are deleting it (see postinit.c), and that no one is
1736 : : * using it as a CREATE DATABASE template or trying to delete it for
1737 : : * themselves.
1738 : : */
1739 : 78 : pgdbrel = table_open(DatabaseRelationId, RowExclusiveLock);
1740 : :
1741 [ + + ]: 78 : if (!get_db_info(dbname, AccessExclusiveLock, &db_id, NULL, NULL,
1742 : : &db_istemplate, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL))
1743 : : {
1744 [ + + ]: 21 : if (!missing_ok)
1745 : : {
1746 [ + - ]: 10 : ereport(ERROR,
1747 : : (errcode(ERRCODE_UNDEFINED_DATABASE),
1748 : : errmsg("database \"%s\" does not exist", dbname)));
1749 : : }
1750 : : else
1751 : : {
1752 : : /* Close pg_database, release the lock, since we changed nothing */
1753 : 11 : table_close(pgdbrel, RowExclusiveLock);
1754 [ + + ]: 11 : ereport(NOTICE,
1755 : : (errmsg("database \"%s\" does not exist, skipping",
1756 : : dbname)));
1757 : 11 : return;
1758 : : }
1759 : : }
1760 : :
1761 : : /*
1762 : : * Permission checks
1763 : : */
1764 [ - + ]: 57 : if (!object_ownercheck(DatabaseRelationId, db_id, GetUserId()))
1765 : 0 : aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_DATABASE,
1766 : : dbname);
1767 : :
1768 : : /* DROP hook for the database being removed */
1769 [ - + ]: 57 : InvokeObjectDropHook(DatabaseRelationId, db_id, 0);
1770 : :
1771 : : /*
1772 : : * Disallow dropping a DB that is marked istemplate. This is just to
1773 : : * prevent people from accidentally dropping template0 or template1; they
1774 : : * can do so if they're really determined ...
1775 : : */
1776 [ - + ]: 57 : if (db_istemplate)
1777 [ # # ]: 0 : ereport(ERROR,
1778 : : (errcode(ERRCODE_WRONG_OBJECT_TYPE),
1779 : : errmsg("cannot drop a template database")));
1780 : :
1781 : : /* Obviously can't drop my own database */
1782 [ - + ]: 57 : if (db_id == MyDatabaseId)
1783 [ # # ]: 0 : ereport(ERROR,
1784 : : (errcode(ERRCODE_OBJECT_IN_USE),
1785 : : errmsg("cannot drop the currently open database")));
1786 : :
1787 : : /*
1788 : : * Check whether there are active logical slots that refer to the
1789 : : * to-be-dropped database. The database lock we are holding prevents the
1790 : : * creation of new slots using the database or existing slots becoming
1791 : : * active.
1792 : : */
1793 : 57 : (void) ReplicationSlotsCountDBSlots(db_id, &nslots, &nslots_active);
1794 [ + + ]: 57 : if (nslots_active)
1795 : : {
1796 [ + - ]: 1 : ereport(ERROR,
1797 : : (errcode(ERRCODE_OBJECT_IN_USE),
1798 : : errmsg("database \"%s\" is used by an active logical replication slot",
1799 : : dbname),
1800 : : errdetail_plural("There is %d active slot.",
1801 : : "There are %d active slots.",
1802 : : nslots_active, nslots_active)));
1803 : : }
1804 : :
1805 : : /*
1806 : : * Check if there are subscriptions defined in the target database.
1807 : : *
1808 : : * We can't drop them automatically because they might be holding
1809 : : * resources in other databases/instances.
1810 : : */
1811 [ - + ]: 56 : if ((nsubscriptions = CountDBSubscriptions(db_id)) > 0)
1812 [ # # ]: 0 : ereport(ERROR,
1813 : : (errcode(ERRCODE_OBJECT_IN_USE),
1814 : : errmsg("database \"%s\" is being used by logical replication subscription",
1815 : : dbname),
1816 : : errdetail_plural("There is %d subscription.",
1817 : : "There are %d subscriptions.",
1818 : : nsubscriptions, nsubscriptions)));
1819 : :
1820 : :
1821 : : /*
1822 : : * Attempt to terminate all existing connections to the target database if
1823 : : * the user has requested to do so.
1824 : : */
1825 [ + + ]: 56 : if (force)
1826 : 2 : TerminateOtherDBBackends(db_id);
1827 : :
1828 : : /*
1829 : : * Check for other backends in the target database. (Because we hold the
1830 : : * database lock, no new ones can start after this.)
1831 : : *
1832 : : * As in CREATE DATABASE, check this after other error conditions.
1833 : : */
1834 [ - + ]: 56 : if (CountOtherDBBackends(db_id, ¬herbackends, &npreparedxacts))
1835 [ # # ]: 0 : ereport(ERROR,
1836 : : (errcode(ERRCODE_OBJECT_IN_USE),
1837 : : errmsg("database \"%s\" is being accessed by other users",
1838 : : dbname),
1839 : : errdetail_busy_db(notherbackends, npreparedxacts)));
1840 : :
1841 : : /*
1842 : : * Delete any comments or security labels associated with the database.
1843 : : */
1844 : 56 : DeleteSharedComments(db_id, DatabaseRelationId);
1845 : 56 : DeleteSharedSecurityLabel(db_id, DatabaseRelationId);
1846 : :
1847 : : /*
1848 : : * Remove settings associated with this database
1849 : : */
1850 : 56 : DropSetting(db_id, InvalidOid);
1851 : :
1852 : : /*
1853 : : * Remove shared dependency references for the database.
1854 : : */
1855 : 56 : dropDatabaseDependencies(db_id);
1856 : :
1857 : : /*
1858 : : * Tell the cumulative stats system to forget it immediately, too.
1859 : : */
1860 : 56 : pgstat_drop_database(db_id);
1861 : :
1862 : : /*
1863 : : * Except for the deletion of the catalog row, subsequent actions are not
1864 : : * transactional (consider DropDatabaseBuffers() discarding modified
1865 : : * buffers). But we might crash or get interrupted below. To prevent
1866 : : * accesses to a database with invalid contents, mark the database as
1867 : : * invalid using an in-place update.
1868 : : *
1869 : : * We need to flush the WAL before continuing, to guarantee the
1870 : : * modification is durable before performing irreversible filesystem
1871 : : * operations.
1872 : : */
1873 : 56 : ScanKeyInit(&scankey,
1874 : : Anum_pg_database_datname,
1875 : : BTEqualStrategyNumber, F_NAMEEQ,
1876 : : CStringGetDatum(dbname));
1877 : 56 : systable_inplace_update_begin(pgdbrel, DatabaseNameIndexId, true,
1878 : : NULL, 1, &scankey, &tup, &inplace_state);
1879 [ - + ]: 56 : if (!HeapTupleIsValid(tup))
1880 [ # # ]: 0 : elog(ERROR, "cache lookup failed for database %u", db_id);
1881 : 56 : datform = (Form_pg_database) GETSTRUCT(tup);
1882 : 56 : datform->datconnlimit = DATCONNLIMIT_INVALID_DB;
1883 : 56 : systable_inplace_update_finish(inplace_state, tup);
1884 : 56 : XLogFlush(XactLastRecEnd);
1885 : :
1886 : 56 : INJECTION_POINT("dropdb-after-invalid-marker", NULL);
1887 : :
1888 : : /*
1889 : : * Also delete the tuple - transactionally. If this transaction commits,
1890 : : * the row will be gone, but if we fail, dropdb() can be invoked again.
1891 : : */
1892 : 55 : CatalogTupleDelete(pgdbrel, &tup->t_self);
1893 : 55 : heap_freetuple(tup);
1894 : :
1895 : : /*
1896 : : * Drop db-specific replication slots.
1897 : : */
1898 : 55 : ReplicationSlotsDropDBSlots(db_id);
1899 : :
1900 : : /*
1901 : : * Drop pages for this database that are in the shared buffer cache. This
1902 : : * is important to ensure that no remaining backend tries to write out a
1903 : : * dirty buffer to the dead database later...
1904 : : */
1905 : 55 : DropDatabaseBuffers(db_id);
1906 : :
1907 : : /*
1908 : : * Tell checkpointer to forget any pending fsync and unlink requests for
1909 : : * files in the database; else the fsyncs will fail at next checkpoint, or
1910 : : * worse, it will delete files that belong to a newly created database
1911 : : * with the same OID.
1912 : : */
1913 : 55 : ForgetDatabaseSyncRequests(db_id);
1914 : :
1915 : : /*
1916 : : * Force a checkpoint to make sure the checkpointer has received the
1917 : : * message sent by ForgetDatabaseSyncRequests.
1918 : : */
1919 : 55 : RequestCheckpoint(CHECKPOINT_FAST | CHECKPOINT_FORCE | CHECKPOINT_WAIT);
1920 : :
1921 : : /* Close all smgr fds in all backends. */
1922 : 55 : WaitForProcSignalBarrier(EmitProcSignalBarrier(PROCSIGNAL_BARRIER_SMGRRELEASE));
1923 : :
1924 : : /*
1925 : : * Remove all tablespace subdirs belonging to the database.
1926 : : */
1927 : 55 : remove_dbtablespaces(db_id);
1928 : :
1929 : : /*
1930 : : * Close pg_database, but keep lock till commit.
1931 : : */
1932 : 55 : table_close(pgdbrel, NoLock);
1933 : :
1934 : : /*
1935 : : * Force synchronous commit, thus minimizing the window between removal of
1936 : : * the database files and committal of the transaction. If we crash before
1937 : : * committing, we'll have a DB that's gone on disk but still there
1938 : : * according to pg_database, which is not good.
1939 : : */
1940 : 55 : ForceSyncCommit();
1941 : : }
1942 : :
1943 : :
1944 : : /*
1945 : : * Rename database
1946 : : */
1947 : : ObjectAddress
1948 : 9 : RenameDatabase(const char *oldname, const char *newname)
1949 : : {
1950 : : Oid db_id;
1951 : : HeapTuple newtup;
1952 : : ItemPointerData otid;
1953 : : Relation rel;
1954 : : int notherbackends;
1955 : : int npreparedxacts;
1956 : : ObjectAddress address;
1957 : :
1958 : : /* Report error if name has \n or \r character. */
1959 [ - + ]: 9 : if (strpbrk(newname, "\n\r"))
1960 [ # # ]: 0 : ereport(ERROR,
1961 : : (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
1962 : : errmsg("database name \"%s\" contains a newline or carriage return character", newname)));
1963 : :
1964 : : /*
1965 : : * Look up the target database's OID, and get exclusive lock on it. We
1966 : : * need this for the same reasons as DROP DATABASE.
1967 : : */
1968 : 9 : rel = table_open(DatabaseRelationId, RowExclusiveLock);
1969 : :
1970 [ - + ]: 9 : if (!get_db_info(oldname, AccessExclusiveLock, &db_id, NULL, NULL, NULL,
1971 : : NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL))
1972 [ # # ]: 0 : ereport(ERROR,
1973 : : (errcode(ERRCODE_UNDEFINED_DATABASE),
1974 : : errmsg("database \"%s\" does not exist", oldname)));
1975 : :
1976 : : /* must be owner */
1977 [ - + ]: 9 : if (!object_ownercheck(DatabaseRelationId, db_id, GetUserId()))
1978 : 0 : aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_DATABASE,
1979 : : oldname);
1980 : :
1981 : : /* must have createdb rights */
1982 [ - + ]: 9 : if (!have_createdb_privilege())
1983 [ # # ]: 0 : ereport(ERROR,
1984 : : (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
1985 : : errmsg("permission denied to rename database")));
1986 : :
1987 : : /*
1988 : : * If built with appropriate switch, whine when regression-testing
1989 : : * conventions for database names are violated.
1990 : : */
1991 : : #ifdef ENFORCE_REGRESSION_TEST_NAME_RESTRICTIONS
1992 : : if (strstr(newname, "regression") == NULL)
1993 : : elog(WARNING, "databases created by regression test cases should have names including \"regression\"");
1994 : : #endif
1995 : :
1996 : : /*
1997 : : * Make sure the new name doesn't exist. See notes for same error in
1998 : : * CREATE DATABASE.
1999 : : */
2000 [ - + ]: 9 : if (OidIsValid(get_database_oid(newname, true)))
2001 [ # # ]: 0 : ereport(ERROR,
2002 : : (errcode(ERRCODE_DUPLICATE_DATABASE),
2003 : : errmsg("database \"%s\" already exists", newname)));
2004 : :
2005 : : /*
2006 : : * XXX Client applications probably store the current database somewhere,
2007 : : * so renaming it could cause confusion. On the other hand, there may not
2008 : : * be an actual problem besides a little confusion, so think about this
2009 : : * and decide.
2010 : : */
2011 [ - + ]: 9 : if (db_id == MyDatabaseId)
2012 [ # # ]: 0 : ereport(ERROR,
2013 : : (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
2014 : : errmsg("current database cannot be renamed")));
2015 : :
2016 : : /*
2017 : : * Make sure the database does not have active sessions. This is the same
2018 : : * concern as above, but applied to other sessions.
2019 : : *
2020 : : * As in CREATE DATABASE, check this after other error conditions.
2021 : : */
2022 [ - + ]: 9 : if (CountOtherDBBackends(db_id, ¬herbackends, &npreparedxacts))
2023 [ # # ]: 0 : ereport(ERROR,
2024 : : (errcode(ERRCODE_OBJECT_IN_USE),
2025 : : errmsg("database \"%s\" is being accessed by other users",
2026 : : oldname),
2027 : : errdetail_busy_db(notherbackends, npreparedxacts)));
2028 : :
2029 : : /* rename */
2030 : 9 : newtup = SearchSysCacheLockedCopy1(DATABASEOID, ObjectIdGetDatum(db_id));
2031 [ - + ]: 9 : if (!HeapTupleIsValid(newtup))
2032 [ # # ]: 0 : elog(ERROR, "cache lookup failed for database %u", db_id);
2033 : 9 : otid = newtup->t_self;
2034 : 9 : namestrcpy(&(((Form_pg_database) GETSTRUCT(newtup))->datname), newname);
2035 : 9 : CatalogTupleUpdate(rel, &otid, newtup);
2036 : 9 : UnlockTuple(rel, &otid, InplaceUpdateTupleLock);
2037 : :
2038 [ - + ]: 9 : InvokeObjectPostAlterHook(DatabaseRelationId, db_id, 0);
2039 : :
2040 : 9 : ObjectAddressSet(address, DatabaseRelationId, db_id);
2041 : :
2042 : : /*
2043 : : * Close pg_database, but keep lock till commit.
2044 : : */
2045 : 9 : table_close(rel, NoLock);
2046 : :
2047 : 9 : return address;
2048 : : }
2049 : :
2050 : :
2051 : : /*
2052 : : * ALTER DATABASE SET TABLESPACE
2053 : : */
2054 : : static void
2055 : 14 : movedb(const char *dbname, const char *tblspcname)
2056 : : {
2057 : : Oid db_id;
2058 : : Relation pgdbrel;
2059 : : int notherbackends;
2060 : : int npreparedxacts;
2061 : : HeapTuple oldtuple,
2062 : : newtuple;
2063 : : Oid src_tblspcoid,
2064 : : dst_tblspcoid;
2065 : : ScanKeyData scankey;
2066 : : SysScanDesc sysscan;
2067 : : AclResult aclresult;
2068 : : char *src_dbpath;
2069 : : char *dst_dbpath;
2070 : : DIR *dstdir;
2071 : : struct dirent *xlde;
2072 : : movedb_failure_params fparms;
2073 : :
2074 : : /*
2075 : : * Look up the target database's OID, and get exclusive lock on it. We
2076 : : * need this to ensure that no new backend starts up in the database while
2077 : : * we are moving it, and that no one is using it as a CREATE DATABASE
2078 : : * template or trying to delete it.
2079 : : */
2080 : 14 : pgdbrel = table_open(DatabaseRelationId, RowExclusiveLock);
2081 : :
2082 [ - + ]: 14 : if (!get_db_info(dbname, AccessExclusiveLock, &db_id, NULL, NULL, NULL,
2083 : : NULL, NULL, NULL, NULL, &src_tblspcoid, NULL, NULL, NULL, NULL, NULL, NULL))
2084 [ # # ]: 0 : ereport(ERROR,
2085 : : (errcode(ERRCODE_UNDEFINED_DATABASE),
2086 : : errmsg("database \"%s\" does not exist", dbname)));
2087 : :
2088 : : /*
2089 : : * We actually need a session lock, so that the lock will persist across
2090 : : * the commit/restart below. (We could almost get away with letting the
2091 : : * lock be released at commit, except that someone could try to move
2092 : : * relations of the DB back into the old directory while we rmtree() it.)
2093 : : */
2094 : 14 : LockSharedObjectForSession(DatabaseRelationId, db_id, 0,
2095 : : AccessExclusiveLock);
2096 : :
2097 : : /*
2098 : : * Permission checks
2099 : : */
2100 [ - + ]: 14 : if (!object_ownercheck(DatabaseRelationId, db_id, GetUserId()))
2101 : 0 : aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_DATABASE,
2102 : : dbname);
2103 : :
2104 : : /*
2105 : : * Obviously can't move the tables of my own database
2106 : : */
2107 [ - + ]: 14 : if (db_id == MyDatabaseId)
2108 [ # # ]: 0 : ereport(ERROR,
2109 : : (errcode(ERRCODE_OBJECT_IN_USE),
2110 : : errmsg("cannot change the tablespace of the currently open database")));
2111 : :
2112 : : /*
2113 : : * Get tablespace's oid
2114 : : */
2115 : 14 : dst_tblspcoid = get_tablespace_oid(tblspcname, false);
2116 : :
2117 : : /*
2118 : : * Permission checks
2119 : : */
2120 : 14 : aclresult = object_aclcheck(TableSpaceRelationId, dst_tblspcoid, GetUserId(),
2121 : : ACL_CREATE);
2122 [ - + ]: 14 : if (aclresult != ACLCHECK_OK)
2123 : 0 : aclcheck_error(aclresult, OBJECT_TABLESPACE,
2124 : : tblspcname);
2125 : :
2126 : : /*
2127 : : * pg_global must never be the default tablespace
2128 : : */
2129 [ - + ]: 14 : if (dst_tblspcoid == GLOBALTABLESPACE_OID)
2130 [ # # ]: 0 : ereport(ERROR,
2131 : : (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
2132 : : errmsg("pg_global cannot be used as default tablespace")));
2133 : :
2134 : : /*
2135 : : * No-op if same tablespace
2136 : : */
2137 [ - + ]: 14 : if (src_tblspcoid == dst_tblspcoid)
2138 : : {
2139 : 0 : table_close(pgdbrel, NoLock);
2140 : 0 : UnlockSharedObjectForSession(DatabaseRelationId, db_id, 0,
2141 : : AccessExclusiveLock);
2142 : 0 : return;
2143 : : }
2144 : :
2145 : : /*
2146 : : * Check for other backends in the target database. (Because we hold the
2147 : : * database lock, no new ones can start after this.)
2148 : : *
2149 : : * As in CREATE DATABASE, check this after other error conditions.
2150 : : */
2151 [ - + ]: 14 : if (CountOtherDBBackends(db_id, ¬herbackends, &npreparedxacts))
2152 [ # # ]: 0 : ereport(ERROR,
2153 : : (errcode(ERRCODE_OBJECT_IN_USE),
2154 : : errmsg("database \"%s\" is being accessed by other users",
2155 : : dbname),
2156 : : errdetail_busy_db(notherbackends, npreparedxacts)));
2157 : :
2158 : : /*
2159 : : * Get old and new database paths
2160 : : */
2161 : 14 : src_dbpath = GetDatabasePath(db_id, src_tblspcoid);
2162 : 14 : dst_dbpath = GetDatabasePath(db_id, dst_tblspcoid);
2163 : :
2164 : : /*
2165 : : * Force a checkpoint before proceeding. This will force all dirty
2166 : : * buffers, including those of unlogged tables, out to disk, to ensure
2167 : : * source database is up-to-date on disk for the copy.
2168 : : * FlushDatabaseBuffers() would suffice for that, but we also want to
2169 : : * process any pending unlink requests. Otherwise, the check for existing
2170 : : * files in the target directory might fail unnecessarily, not to mention
2171 : : * that the copy might fail due to source files getting deleted under it.
2172 : : * On Windows, this also ensures that background procs don't hold any open
2173 : : * files, which would cause rmdir() to fail.
2174 : : */
2175 : 14 : RequestCheckpoint(CHECKPOINT_FAST | CHECKPOINT_FORCE | CHECKPOINT_WAIT
2176 : : | CHECKPOINT_FLUSH_UNLOGGED);
2177 : :
2178 : : /* Close all smgr fds in all backends. */
2179 : 14 : WaitForProcSignalBarrier(EmitProcSignalBarrier(PROCSIGNAL_BARRIER_SMGRRELEASE));
2180 : :
2181 : : /*
2182 : : * Now drop all buffers holding data of the target database; they should
2183 : : * no longer be dirty so DropDatabaseBuffers is safe.
2184 : : *
2185 : : * It might seem that we could just let these buffers age out of shared
2186 : : * buffers naturally, since they should not get referenced anymore. The
2187 : : * problem with that is that if the user later moves the database back to
2188 : : * its original tablespace, any still-surviving buffers would appear to
2189 : : * contain valid data again --- but they'd be missing any changes made in
2190 : : * the database while it was in the new tablespace. In any case, freeing
2191 : : * buffers that should never be used again seems worth the cycles.
2192 : : *
2193 : : * Note: it'd be sufficient to get rid of buffers matching db_id and
2194 : : * src_tblspcoid, but bufmgr.c presently provides no API for that.
2195 : : */
2196 : 14 : DropDatabaseBuffers(db_id);
2197 : :
2198 : : /*
2199 : : * Check for existence of files in the target directory, i.e., objects of
2200 : : * this database that are already in the target tablespace. We can't
2201 : : * allow the move in such a case, because we would need to change those
2202 : : * relations' pg_class.reltablespace entries to zero, and we don't have
2203 : : * access to the DB's pg_class to do so.
2204 : : */
2205 : 14 : dstdir = AllocateDir(dst_dbpath);
2206 [ - + ]: 14 : if (dstdir != NULL)
2207 : : {
2208 [ # # ]: 0 : while ((xlde = ReadDir(dstdir, dst_dbpath)) != NULL)
2209 : : {
2210 [ # # ]: 0 : if (strcmp(xlde->d_name, ".") == 0 ||
2211 [ # # ]: 0 : strcmp(xlde->d_name, "..") == 0)
2212 : 0 : continue;
2213 : :
2214 [ # # ]: 0 : ereport(ERROR,
2215 : : (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
2216 : : errmsg("some relations of database \"%s\" are already in tablespace \"%s\"",
2217 : : dbname, tblspcname),
2218 : : errhint("You must move them back to the database's default tablespace before using this command.")));
2219 : : }
2220 : :
2221 : 0 : FreeDir(dstdir);
2222 : :
2223 : : /*
2224 : : * The directory exists but is empty. We must remove it before using
2225 : : * the copydir function.
2226 : : */
2227 [ # # ]: 0 : if (rmdir(dst_dbpath) != 0)
2228 [ # # ]: 0 : elog(ERROR, "could not remove directory \"%s\": %m",
2229 : : dst_dbpath);
2230 : : }
2231 : :
2232 : : /*
2233 : : * Use an ENSURE block to make sure we remove the debris if the copy fails
2234 : : * (eg, due to out-of-disk-space). This is not a 100% solution, because
2235 : : * of the possibility of failure during transaction commit, but it should
2236 : : * handle most scenarios.
2237 : : */
2238 : 14 : fparms.dest_dboid = db_id;
2239 : 14 : fparms.dest_tsoid = dst_tblspcoid;
2240 [ + - ]: 14 : PG_ENSURE_ERROR_CLEANUP(movedb_failure_callback,
2241 : : PointerGetDatum(&fparms));
2242 : : {
2243 : 14 : Datum new_record[Natts_pg_database] = {0};
2244 : 14 : bool new_record_nulls[Natts_pg_database] = {0};
2245 : 14 : bool new_record_repl[Natts_pg_database] = {0};
2246 : :
2247 : : /*
2248 : : * Copy files from the old tablespace to the new one
2249 : : */
2250 : 14 : copydir(src_dbpath, dst_dbpath, false);
2251 : :
2252 : : /*
2253 : : * Record the filesystem change in XLOG
2254 : : */
2255 : : {
2256 : : xl_dbase_create_file_copy_rec xlrec;
2257 : :
2258 : 14 : xlrec.db_id = db_id;
2259 : 14 : xlrec.tablespace_id = dst_tblspcoid;
2260 : 14 : xlrec.src_db_id = db_id;
2261 : 14 : xlrec.src_tablespace_id = src_tblspcoid;
2262 : :
2263 : 14 : XLogBeginInsert();
2264 : 14 : XLogRegisterData(&xlrec,
2265 : : sizeof(xl_dbase_create_file_copy_rec));
2266 : :
2267 : 14 : (void) XLogInsert(RM_DBASE_ID,
2268 : : XLOG_DBASE_CREATE_FILE_COPY | XLR_SPECIAL_REL_UPDATE);
2269 : : }
2270 : :
2271 : : /*
2272 : : * Update the database's pg_database tuple
2273 : : */
2274 : 14 : ScanKeyInit(&scankey,
2275 : : Anum_pg_database_datname,
2276 : : BTEqualStrategyNumber, F_NAMEEQ,
2277 : : CStringGetDatum(dbname));
2278 : 14 : sysscan = systable_beginscan(pgdbrel, DatabaseNameIndexId, true,
2279 : : NULL, 1, &scankey);
2280 : 14 : oldtuple = systable_getnext(sysscan);
2281 [ - + ]: 14 : if (!HeapTupleIsValid(oldtuple)) /* shouldn't happen... */
2282 [ # # ]: 0 : ereport(ERROR,
2283 : : (errcode(ERRCODE_UNDEFINED_DATABASE),
2284 : : errmsg("database \"%s\" does not exist", dbname)));
2285 : 14 : LockTuple(pgdbrel, &oldtuple->t_self, InplaceUpdateTupleLock);
2286 : :
2287 : 14 : new_record[Anum_pg_database_dattablespace - 1] = ObjectIdGetDatum(dst_tblspcoid);
2288 : 14 : new_record_repl[Anum_pg_database_dattablespace - 1] = true;
2289 : :
2290 : 14 : newtuple = heap_modify_tuple(oldtuple, RelationGetDescr(pgdbrel),
2291 : : new_record,
2292 : : new_record_nulls, new_record_repl);
2293 : 14 : CatalogTupleUpdate(pgdbrel, &oldtuple->t_self, newtuple);
2294 : 14 : UnlockTuple(pgdbrel, &oldtuple->t_self, InplaceUpdateTupleLock);
2295 : :
2296 [ - + ]: 14 : InvokeObjectPostAlterHook(DatabaseRelationId, db_id, 0);
2297 : :
2298 : 14 : systable_endscan(sysscan);
2299 : :
2300 : : /*
2301 : : * Force another checkpoint here. As in CREATE DATABASE, this is to
2302 : : * ensure that we don't have to replay a committed
2303 : : * XLOG_DBASE_CREATE_FILE_COPY operation, which would cause us to lose
2304 : : * any unlogged operations done in the new DB tablespace before the
2305 : : * next checkpoint.
2306 : : */
2307 : 14 : RequestCheckpoint(CHECKPOINT_FAST | CHECKPOINT_FORCE | CHECKPOINT_WAIT);
2308 : :
2309 : : /*
2310 : : * Force synchronous commit, thus minimizing the window between
2311 : : * copying the database files and committal of the transaction. If we
2312 : : * crash before committing, we'll leave an orphaned set of files on
2313 : : * disk, which is not fatal but not good either.
2314 : : */
2315 : 14 : ForceSyncCommit();
2316 : :
2317 : : /*
2318 : : * Close pg_database, but keep lock till commit.
2319 : : */
2320 : 14 : table_close(pgdbrel, NoLock);
2321 : : }
2322 [ - + ]: 14 : PG_END_ENSURE_ERROR_CLEANUP(movedb_failure_callback,
2323 : : PointerGetDatum(&fparms));
2324 : :
2325 : : /*
2326 : : * Commit the transaction so that the pg_database update is committed. If
2327 : : * we crash while removing files, the database won't be corrupt, we'll
2328 : : * just leave some orphaned files in the old directory.
2329 : : *
2330 : : * (This is OK because we know we aren't inside a transaction block.)
2331 : : *
2332 : : * XXX would it be safe/better to do this inside the ensure block? Not
2333 : : * convinced it's a good idea; consider elog just after the transaction
2334 : : * really commits.
2335 : : */
2336 : 14 : PopActiveSnapshot();
2337 : 14 : CommitTransactionCommand();
2338 : :
2339 : : /* Start new transaction for the remaining work; don't need a snapshot */
2340 : 14 : StartTransactionCommand();
2341 : :
2342 : : /*
2343 : : * Remove files from the old tablespace
2344 : : */
2345 [ - + ]: 14 : if (!rmtree(src_dbpath, true))
2346 [ # # ]: 0 : ereport(WARNING,
2347 : : (errmsg("some useless files may be left behind in old database directory \"%s\"",
2348 : : src_dbpath)));
2349 : :
2350 : : /*
2351 : : * Record the filesystem change in XLOG
2352 : : */
2353 : : {
2354 : : xl_dbase_drop_rec xlrec;
2355 : :
2356 : 14 : xlrec.db_id = db_id;
2357 : 14 : xlrec.ntablespaces = 1;
2358 : :
2359 : 14 : XLogBeginInsert();
2360 : 14 : XLogRegisterData(&xlrec, sizeof(xl_dbase_drop_rec));
2361 : 14 : XLogRegisterData(&src_tblspcoid, sizeof(Oid));
2362 : :
2363 : 14 : (void) XLogInsert(RM_DBASE_ID,
2364 : : XLOG_DBASE_DROP | XLR_SPECIAL_REL_UPDATE);
2365 : : }
2366 : :
2367 : : /* Now it's safe to release the database lock */
2368 : 14 : UnlockSharedObjectForSession(DatabaseRelationId, db_id, 0,
2369 : : AccessExclusiveLock);
2370 : :
2371 : 14 : pfree(src_dbpath);
2372 : 14 : pfree(dst_dbpath);
2373 : : }
2374 : :
2375 : : /* Error cleanup callback for movedb */
2376 : : static void
2377 : 0 : movedb_failure_callback(int code, Datum arg)
2378 : : {
2379 : 0 : movedb_failure_params *fparms = (movedb_failure_params *) DatumGetPointer(arg);
2380 : : char *dstpath;
2381 : :
2382 : : /* Get rid of anything we managed to copy to the target directory */
2383 : 0 : dstpath = GetDatabasePath(fparms->dest_dboid, fparms->dest_tsoid);
2384 : :
2385 : 0 : (void) rmtree(dstpath, true);
2386 : :
2387 : 0 : pfree(dstpath);
2388 : 0 : }
2389 : :
2390 : : /*
2391 : : * Process options and call dropdb function.
2392 : : */
2393 : : void
2394 : 78 : DropDatabase(ParseState *pstate, DropdbStmt *stmt)
2395 : : {
2396 : 78 : bool force = false;
2397 : : ListCell *lc;
2398 : :
2399 [ + + + + : 96 : foreach(lc, stmt->options)
+ + ]
2400 : : {
2401 : 18 : DefElem *opt = (DefElem *) lfirst(lc);
2402 : :
2403 [ + - ]: 18 : if (strcmp(opt->defname, "force") == 0)
2404 : 18 : force = true;
2405 : : else
2406 [ # # ]: 0 : ereport(ERROR,
2407 : : (errcode(ERRCODE_SYNTAX_ERROR),
2408 : : errmsg("unrecognized %s option \"%s\"",
2409 : : "DROP DATABASE", opt->defname),
2410 : : parser_errposition(pstate, opt->location)));
2411 : : }
2412 : :
2413 : 78 : dropdb(stmt->dbname, stmt->missing_ok, force);
2414 : 66 : }
2415 : :
2416 : : /*
2417 : : * ALTER DATABASE name ...
2418 : : */
2419 : : Oid
2420 : 53 : AlterDatabase(ParseState *pstate, AlterDatabaseStmt *stmt, bool isTopLevel)
2421 : : {
2422 : : Relation rel;
2423 : : Oid dboid;
2424 : : HeapTuple tuple,
2425 : : newtuple;
2426 : : Form_pg_database datform;
2427 : : ScanKeyData scankey;
2428 : : SysScanDesc scan;
2429 : : ListCell *option;
2430 : 53 : bool dbistemplate = false;
2431 : 53 : bool dballowconnections = true;
2432 : 53 : int dbconnlimit = DATCONNLIMIT_UNLIMITED;
2433 : 53 : DefElem *distemplate = NULL;
2434 : 53 : DefElem *dallowconnections = NULL;
2435 : 53 : DefElem *dconnlimit = NULL;
2436 : 53 : DefElem *dtablespace = NULL;
2437 : 53 : Datum new_record[Natts_pg_database] = {0};
2438 : 53 : bool new_record_nulls[Natts_pg_database] = {0};
2439 : 53 : bool new_record_repl[Natts_pg_database] = {0};
2440 : :
2441 : : /* Extract options from the statement node tree */
2442 [ + - + + : 106 : foreach(option, stmt->options)
+ + ]
2443 : : {
2444 : 53 : DefElem *defel = (DefElem *) lfirst(option);
2445 : :
2446 [ + + ]: 53 : if (strcmp(defel->defname, "is_template") == 0)
2447 : : {
2448 [ - + ]: 12 : if (distemplate)
2449 : 0 : errorConflictingDefElem(defel, pstate);
2450 : 12 : distemplate = defel;
2451 : : }
2452 [ + + ]: 41 : else if (strcmp(defel->defname, "allow_connections") == 0)
2453 : : {
2454 [ - + ]: 21 : if (dallowconnections)
2455 : 0 : errorConflictingDefElem(defel, pstate);
2456 : 21 : dallowconnections = defel;
2457 : : }
2458 [ + + ]: 20 : else if (strcmp(defel->defname, "connection_limit") == 0)
2459 : : {
2460 [ - + ]: 6 : if (dconnlimit)
2461 : 0 : errorConflictingDefElem(defel, pstate);
2462 : 6 : dconnlimit = defel;
2463 : : }
2464 [ + - ]: 14 : else if (strcmp(defel->defname, "tablespace") == 0)
2465 : : {
2466 [ - + ]: 14 : if (dtablespace)
2467 : 0 : errorConflictingDefElem(defel, pstate);
2468 : 14 : dtablespace = defel;
2469 : : }
2470 : : else
2471 [ # # ]: 0 : ereport(ERROR,
2472 : : (errcode(ERRCODE_SYNTAX_ERROR),
2473 : : errmsg("option \"%s\" not recognized", defel->defname),
2474 : : parser_errposition(pstate, defel->location)));
2475 : : }
2476 : :
2477 [ + + ]: 53 : if (dtablespace)
2478 : : {
2479 : : /*
2480 : : * While the SET TABLESPACE syntax doesn't allow any other options,
2481 : : * somebody could write "WITH TABLESPACE ...". Forbid any other
2482 : : * options from being specified in that case.
2483 : : */
2484 [ - + ]: 14 : if (list_length(stmt->options) != 1)
2485 [ # # ]: 0 : ereport(ERROR,
2486 : : (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
2487 : : errmsg("option \"%s\" cannot be specified with other options",
2488 : : dtablespace->defname),
2489 : : parser_errposition(pstate, dtablespace->location)));
2490 : : /* this case isn't allowed within a transaction block */
2491 : 14 : PreventInTransactionBlock(isTopLevel, "ALTER DATABASE SET TABLESPACE");
2492 : 14 : movedb(stmt->dbname, defGetString(dtablespace));
2493 : 14 : return InvalidOid;
2494 : : }
2495 : :
2496 [ + + + - ]: 39 : if (distemplate && distemplate->arg)
2497 : 12 : dbistemplate = defGetBoolean(distemplate);
2498 [ + + + - ]: 39 : if (dallowconnections && dallowconnections->arg)
2499 : 21 : dballowconnections = defGetBoolean(dallowconnections);
2500 [ + + + - ]: 39 : if (dconnlimit && dconnlimit->arg)
2501 : : {
2502 : 6 : dbconnlimit = defGetInt32(dconnlimit);
2503 [ - + ]: 6 : if (dbconnlimit < DATCONNLIMIT_UNLIMITED)
2504 [ # # ]: 0 : ereport(ERROR,
2505 : : (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
2506 : : errmsg("invalid connection limit: %d", dbconnlimit)));
2507 : : }
2508 : :
2509 : : /*
2510 : : * Get the old tuple. We don't need a lock on the database per se,
2511 : : * because we're not going to do anything that would mess up incoming
2512 : : * connections.
2513 : : */
2514 : 39 : rel = table_open(DatabaseRelationId, RowExclusiveLock);
2515 : 39 : ScanKeyInit(&scankey,
2516 : : Anum_pg_database_datname,
2517 : : BTEqualStrategyNumber, F_NAMEEQ,
2518 : 39 : CStringGetDatum(stmt->dbname));
2519 : 39 : scan = systable_beginscan(rel, DatabaseNameIndexId, true,
2520 : : NULL, 1, &scankey);
2521 : 39 : tuple = systable_getnext(scan);
2522 [ - + ]: 39 : if (!HeapTupleIsValid(tuple))
2523 [ # # ]: 0 : ereport(ERROR,
2524 : : (errcode(ERRCODE_UNDEFINED_DATABASE),
2525 : : errmsg("database \"%s\" does not exist", stmt->dbname)));
2526 : 39 : LockTuple(rel, &tuple->t_self, InplaceUpdateTupleLock);
2527 : :
2528 : 39 : datform = (Form_pg_database) GETSTRUCT(tuple);
2529 : 39 : dboid = datform->oid;
2530 : :
2531 [ + + ]: 39 : if (database_is_invalid_form(datform))
2532 : : {
2533 [ + - ]: 1 : ereport(FATAL,
2534 : : errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
2535 : : errmsg("cannot alter invalid database \"%s\"", stmt->dbname),
2536 : : errhint("Use DROP DATABASE to drop invalid databases."));
2537 : : }
2538 : :
2539 [ - + ]: 38 : if (!object_ownercheck(DatabaseRelationId, dboid, GetUserId()))
2540 : 0 : aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_DATABASE,
2541 : 0 : stmt->dbname);
2542 : :
2543 : : /*
2544 : : * In order to avoid getting locked out and having to go through
2545 : : * standalone mode, we refuse to disallow connections to the database
2546 : : * we're currently connected to. Lockout can still happen with concurrent
2547 : : * sessions but the likeliness of that is not high enough to worry about.
2548 : : */
2549 [ + + - + ]: 38 : if (!dballowconnections && dboid == MyDatabaseId)
2550 [ # # ]: 0 : ereport(ERROR,
2551 : : (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
2552 : : errmsg("cannot disallow connections for current database")));
2553 : :
2554 : : /*
2555 : : * Build an updated tuple, perusing the information just obtained
2556 : : */
2557 [ + + ]: 38 : if (distemplate)
2558 : : {
2559 : 12 : new_record[Anum_pg_database_datistemplate - 1] = BoolGetDatum(dbistemplate);
2560 : 12 : new_record_repl[Anum_pg_database_datistemplate - 1] = true;
2561 : : }
2562 [ + + ]: 38 : if (dallowconnections)
2563 : : {
2564 : 21 : new_record[Anum_pg_database_datallowconn - 1] = BoolGetDatum(dballowconnections);
2565 : 21 : new_record_repl[Anum_pg_database_datallowconn - 1] = true;
2566 : : }
2567 [ + + ]: 38 : if (dconnlimit)
2568 : : {
2569 : 5 : new_record[Anum_pg_database_datconnlimit - 1] = Int32GetDatum(dbconnlimit);
2570 : 5 : new_record_repl[Anum_pg_database_datconnlimit - 1] = true;
2571 : : }
2572 : :
2573 : 38 : newtuple = heap_modify_tuple(tuple, RelationGetDescr(rel), new_record,
2574 : : new_record_nulls, new_record_repl);
2575 : 38 : CatalogTupleUpdate(rel, &tuple->t_self, newtuple);
2576 : 38 : UnlockTuple(rel, &tuple->t_self, InplaceUpdateTupleLock);
2577 : :
2578 [ - + ]: 38 : InvokeObjectPostAlterHook(DatabaseRelationId, dboid, 0);
2579 : :
2580 : 38 : systable_endscan(scan);
2581 : :
2582 : : /* Close pg_database, but keep lock till commit */
2583 : 38 : table_close(rel, NoLock);
2584 : :
2585 : 38 : return dboid;
2586 : : }
2587 : :
2588 : :
2589 : : /*
2590 : : * ALTER DATABASE name REFRESH COLLATION VERSION
2591 : : */
2592 : : ObjectAddress
2593 : 4 : AlterDatabaseRefreshColl(AlterDatabaseRefreshCollStmt *stmt)
2594 : : {
2595 : : Relation rel;
2596 : : ScanKeyData scankey;
2597 : : SysScanDesc scan;
2598 : : Oid db_id;
2599 : : HeapTuple tuple;
2600 : : Form_pg_database datForm;
2601 : : ObjectAddress address;
2602 : : Datum datum;
2603 : : bool isnull;
2604 : : char *oldversion;
2605 : : char *newversion;
2606 : :
2607 : 4 : rel = table_open(DatabaseRelationId, RowExclusiveLock);
2608 : 4 : ScanKeyInit(&scankey,
2609 : : Anum_pg_database_datname,
2610 : : BTEqualStrategyNumber, F_NAMEEQ,
2611 : 4 : CStringGetDatum(stmt->dbname));
2612 : 4 : scan = systable_beginscan(rel, DatabaseNameIndexId, true,
2613 : : NULL, 1, &scankey);
2614 : 4 : tuple = systable_getnext(scan);
2615 [ - + ]: 4 : if (!HeapTupleIsValid(tuple))
2616 [ # # ]: 0 : ereport(ERROR,
2617 : : (errcode(ERRCODE_UNDEFINED_DATABASE),
2618 : : errmsg("database \"%s\" does not exist", stmt->dbname)));
2619 : :
2620 : 4 : datForm = (Form_pg_database) GETSTRUCT(tuple);
2621 : 4 : db_id = datForm->oid;
2622 : :
2623 [ - + ]: 4 : if (!object_ownercheck(DatabaseRelationId, db_id, GetUserId()))
2624 : 0 : aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_DATABASE,
2625 : 0 : stmt->dbname);
2626 : 4 : LockTuple(rel, &tuple->t_self, InplaceUpdateTupleLock);
2627 : :
2628 : 4 : datum = heap_getattr(tuple, Anum_pg_database_datcollversion, RelationGetDescr(rel), &isnull);
2629 [ + - ]: 4 : oldversion = isnull ? NULL : TextDatumGetCString(datum);
2630 : :
2631 [ + + ]: 4 : if (datForm->datlocprovider == COLLPROVIDER_LIBC)
2632 : : {
2633 : 3 : datum = heap_getattr(tuple, Anum_pg_database_datcollate, RelationGetDescr(rel), &isnull);
2634 [ - + ]: 3 : if (isnull)
2635 [ # # ]: 0 : elog(ERROR, "unexpected null in pg_database");
2636 : : }
2637 : : else
2638 : : {
2639 : 1 : datum = heap_getattr(tuple, Anum_pg_database_datlocale, RelationGetDescr(rel), &isnull);
2640 [ - + ]: 1 : if (isnull)
2641 [ # # ]: 0 : elog(ERROR, "unexpected null in pg_database");
2642 : : }
2643 : :
2644 : 4 : newversion = get_collation_actual_version(datForm->datlocprovider,
2645 : 4 : TextDatumGetCString(datum));
2646 : :
2647 : : /* cannot change from NULL to non-NULL or vice versa */
2648 [ - + - - : 4 : if ((!oldversion && newversion) || (oldversion && !newversion))
+ - - + ]
2649 [ # # ]: 0 : elog(ERROR, "invalid collation version change");
2650 [ + - + - : 4 : else if (oldversion && newversion && strcmp(newversion, oldversion) != 0)
- + ]
2651 : 0 : {
2652 : 0 : bool nulls[Natts_pg_database] = {0};
2653 : 0 : bool replaces[Natts_pg_database] = {0};
2654 : 0 : Datum values[Natts_pg_database] = {0};
2655 : : HeapTuple newtuple;
2656 : :
2657 [ # # ]: 0 : ereport(NOTICE,
2658 : : (errmsg("changing version from %s to %s",
2659 : : oldversion, newversion)));
2660 : :
2661 : 0 : values[Anum_pg_database_datcollversion - 1] = CStringGetTextDatum(newversion);
2662 : 0 : replaces[Anum_pg_database_datcollversion - 1] = true;
2663 : :
2664 : 0 : newtuple = heap_modify_tuple(tuple, RelationGetDescr(rel),
2665 : : values, nulls, replaces);
2666 : 0 : CatalogTupleUpdate(rel, &tuple->t_self, newtuple);
2667 : 0 : heap_freetuple(newtuple);
2668 : : }
2669 : : else
2670 [ + - ]: 4 : ereport(NOTICE,
2671 : : (errmsg("version has not changed")));
2672 : 4 : UnlockTuple(rel, &tuple->t_self, InplaceUpdateTupleLock);
2673 : :
2674 [ - + ]: 4 : InvokeObjectPostAlterHook(DatabaseRelationId, db_id, 0);
2675 : :
2676 : 4 : ObjectAddressSet(address, DatabaseRelationId, db_id);
2677 : :
2678 : 4 : systable_endscan(scan);
2679 : :
2680 : 4 : table_close(rel, NoLock);
2681 : :
2682 : 4 : return address;
2683 : : }
2684 : :
2685 : :
2686 : : /*
2687 : : * ALTER DATABASE name SET ...
2688 : : */
2689 : : Oid
2690 : 676 : AlterDatabaseSet(AlterDatabaseSetStmt *stmt)
2691 : : {
2692 : 676 : Oid datid = get_database_oid(stmt->dbname, false);
2693 : :
2694 : : /*
2695 : : * Obtain a lock on the database and make sure it didn't go away in the
2696 : : * meantime.
2697 : : */
2698 : 676 : shdepLockAndCheckObject(DatabaseRelationId, datid);
2699 : :
2700 [ - + ]: 676 : if (!object_ownercheck(DatabaseRelationId, datid, GetUserId()))
2701 : 0 : aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_DATABASE,
2702 : 0 : stmt->dbname);
2703 : :
2704 : 676 : AlterSetting(datid, InvalidOid, stmt->setstmt);
2705 : :
2706 : 674 : UnlockSharedObject(DatabaseRelationId, datid, 0, AccessShareLock);
2707 : :
2708 : 674 : return datid;
2709 : : }
2710 : :
2711 : :
2712 : : /*
2713 : : * ALTER DATABASE name OWNER TO newowner
2714 : : */
2715 : : ObjectAddress
2716 : 50 : AlterDatabaseOwner(const char *dbname, Oid newOwnerId)
2717 : : {
2718 : : Oid db_id;
2719 : : HeapTuple tuple;
2720 : : Relation rel;
2721 : : ScanKeyData scankey;
2722 : : SysScanDesc scan;
2723 : : Form_pg_database datForm;
2724 : : ObjectAddress address;
2725 : :
2726 : : /*
2727 : : * Get the old tuple. We don't need a lock on the database per se,
2728 : : * because we're not going to do anything that would mess up incoming
2729 : : * connections.
2730 : : */
2731 : 50 : rel = table_open(DatabaseRelationId, RowExclusiveLock);
2732 : 50 : ScanKeyInit(&scankey,
2733 : : Anum_pg_database_datname,
2734 : : BTEqualStrategyNumber, F_NAMEEQ,
2735 : : CStringGetDatum(dbname));
2736 : 50 : scan = systable_beginscan(rel, DatabaseNameIndexId, true,
2737 : : NULL, 1, &scankey);
2738 : 50 : tuple = systable_getnext(scan);
2739 [ - + ]: 50 : if (!HeapTupleIsValid(tuple))
2740 [ # # ]: 0 : ereport(ERROR,
2741 : : (errcode(ERRCODE_UNDEFINED_DATABASE),
2742 : : errmsg("database \"%s\" does not exist", dbname)));
2743 : :
2744 : 50 : datForm = (Form_pg_database) GETSTRUCT(tuple);
2745 : 50 : db_id = datForm->oid;
2746 : :
2747 : : /*
2748 : : * If the new owner is the same as the existing owner, consider the
2749 : : * command to have succeeded. This is to be consistent with other
2750 : : * objects.
2751 : : */
2752 [ + + ]: 50 : if (datForm->datdba != newOwnerId)
2753 : : {
2754 : : Datum repl_val[Natts_pg_database];
2755 : 18 : bool repl_null[Natts_pg_database] = {0};
2756 : 18 : bool repl_repl[Natts_pg_database] = {0};
2757 : : Acl *newAcl;
2758 : : Datum aclDatum;
2759 : : bool isNull;
2760 : : HeapTuple newtuple;
2761 : :
2762 : : /* Otherwise, must be owner of the existing object */
2763 [ - + ]: 18 : if (!object_ownercheck(DatabaseRelationId, db_id, GetUserId()))
2764 : 0 : aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_DATABASE,
2765 : : dbname);
2766 : :
2767 : : /* Must be able to become new owner */
2768 : 18 : check_can_set_role(GetUserId(), newOwnerId);
2769 : :
2770 : : /*
2771 : : * must have createdb rights
2772 : : *
2773 : : * NOTE: This is different from other alter-owner checks in that the
2774 : : * current user is checked for createdb privileges instead of the
2775 : : * destination owner. This is consistent with the CREATE case for
2776 : : * databases. Because superusers will always have this right, we need
2777 : : * no special case for them.
2778 : : */
2779 [ - + ]: 18 : if (!have_createdb_privilege())
2780 [ # # ]: 0 : ereport(ERROR,
2781 : : (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
2782 : : errmsg("permission denied to change owner of database")));
2783 : :
2784 : 18 : LockTuple(rel, &tuple->t_self, InplaceUpdateTupleLock);
2785 : :
2786 : 18 : repl_repl[Anum_pg_database_datdba - 1] = true;
2787 : 18 : repl_val[Anum_pg_database_datdba - 1] = ObjectIdGetDatum(newOwnerId);
2788 : :
2789 : : /*
2790 : : * Determine the modified ACL for the new owner. This is only
2791 : : * necessary when the ACL is non-null.
2792 : : */
2793 : 18 : aclDatum = heap_getattr(tuple,
2794 : : Anum_pg_database_datacl,
2795 : : RelationGetDescr(rel),
2796 : : &isNull);
2797 [ - + ]: 18 : if (!isNull)
2798 : : {
2799 : 0 : newAcl = aclnewowner(DatumGetAclP(aclDatum),
2800 : : datForm->datdba, newOwnerId);
2801 : 0 : repl_repl[Anum_pg_database_datacl - 1] = true;
2802 : 0 : repl_val[Anum_pg_database_datacl - 1] = PointerGetDatum(newAcl);
2803 : : }
2804 : :
2805 : 18 : newtuple = heap_modify_tuple(tuple, RelationGetDescr(rel), repl_val, repl_null, repl_repl);
2806 : 18 : CatalogTupleUpdate(rel, &newtuple->t_self, newtuple);
2807 : 18 : UnlockTuple(rel, &tuple->t_self, InplaceUpdateTupleLock);
2808 : :
2809 : 18 : heap_freetuple(newtuple);
2810 : :
2811 : : /* Update owner dependency reference */
2812 : 18 : changeDependencyOnOwner(DatabaseRelationId, db_id, newOwnerId);
2813 : : }
2814 : :
2815 [ - + ]: 50 : InvokeObjectPostAlterHook(DatabaseRelationId, db_id, 0);
2816 : :
2817 : 50 : ObjectAddressSet(address, DatabaseRelationId, db_id);
2818 : :
2819 : 50 : systable_endscan(scan);
2820 : :
2821 : : /* Close pg_database, but keep lock till commit */
2822 : 50 : table_close(rel, NoLock);
2823 : :
2824 : 50 : return address;
2825 : : }
2826 : :
2827 : :
2828 : : Datum
2829 : 55 : pg_database_collation_actual_version(PG_FUNCTION_ARGS)
2830 : : {
2831 : 55 : Oid dbid = PG_GETARG_OID(0);
2832 : : HeapTuple tp;
2833 : : char datlocprovider;
2834 : : Datum datum;
2835 : : char *version;
2836 : :
2837 : 55 : tp = SearchSysCache1(DATABASEOID, ObjectIdGetDatum(dbid));
2838 [ - + ]: 55 : if (!HeapTupleIsValid(tp))
2839 [ # # ]: 0 : ereport(ERROR,
2840 : : (errcode(ERRCODE_UNDEFINED_OBJECT),
2841 : : errmsg("database with OID %u does not exist", dbid)));
2842 : :
2843 : 55 : datlocprovider = ((Form_pg_database) GETSTRUCT(tp))->datlocprovider;
2844 : :
2845 [ + + ]: 55 : if (datlocprovider == COLLPROVIDER_LIBC)
2846 : 48 : datum = SysCacheGetAttrNotNull(DATABASEOID, tp, Anum_pg_database_datcollate);
2847 : : else
2848 : 7 : datum = SysCacheGetAttrNotNull(DATABASEOID, tp, Anum_pg_database_datlocale);
2849 : :
2850 : 55 : version = get_collation_actual_version(datlocprovider,
2851 : 55 : TextDatumGetCString(datum));
2852 : :
2853 : 55 : ReleaseSysCache(tp);
2854 : :
2855 [ + + ]: 55 : if (version)
2856 : 41 : PG_RETURN_TEXT_P(cstring_to_text(version));
2857 : : else
2858 : 14 : PG_RETURN_NULL();
2859 : : }
2860 : :
2861 : :
2862 : : /*
2863 : : * Helper functions
2864 : : */
2865 : :
2866 : : /*
2867 : : * Look up info about the database named "name". If the database exists,
2868 : : * obtain the specified lock type on it, fill in any of the remaining
2869 : : * parameters that aren't NULL, and return true. If no such database,
2870 : : * return false.
2871 : : */
2872 : : static bool
2873 : 541 : get_db_info(const char *name, LOCKMODE lockmode,
2874 : : Oid *dbIdP, Oid *ownerIdP,
2875 : : int *encodingP, bool *dbIsTemplateP, bool *dbAllowConnP, bool *dbHasLoginEvtP,
2876 : : TransactionId *dbFrozenXidP, MultiXactId *dbMinMultiP,
2877 : : Oid *dbTablespace, char **dbCollate, char **dbCtype, char **dbLocale,
2878 : : char **dbIcurules,
2879 : : char *dbLocProvider,
2880 : : char **dbCollversion)
2881 : : {
2882 : 541 : bool result = false;
2883 : : Relation relation;
2884 : :
2885 : : Assert(name);
2886 : :
2887 : : /* Caller may wish to grab a better lock on pg_database beforehand... */
2888 : 541 : relation = table_open(DatabaseRelationId, AccessShareLock);
2889 : :
2890 : : /*
2891 : : * Loop covers the rare case where the database is renamed before we can
2892 : : * lock it. We try again just in case we can find a new one of the same
2893 : : * name.
2894 : : */
2895 : : for (;;)
2896 : 0 : {
2897 : : ScanKeyData scanKey;
2898 : : SysScanDesc scan;
2899 : : HeapTuple tuple;
2900 : : Oid dbOid;
2901 : :
2902 : : /*
2903 : : * there's no syscache for database-indexed-by-name, so must do it the
2904 : : * hard way
2905 : : */
2906 : 541 : ScanKeyInit(&scanKey,
2907 : : Anum_pg_database_datname,
2908 : : BTEqualStrategyNumber, F_NAMEEQ,
2909 : : CStringGetDatum(name));
2910 : :
2911 : 541 : scan = systable_beginscan(relation, DatabaseNameIndexId, true,
2912 : : NULL, 1, &scanKey);
2913 : :
2914 : 541 : tuple = systable_getnext(scan);
2915 : :
2916 [ + + ]: 541 : if (!HeapTupleIsValid(tuple))
2917 : : {
2918 : : /* definitely no database of that name */
2919 : 21 : systable_endscan(scan);
2920 : 21 : break;
2921 : : }
2922 : :
2923 : 520 : dbOid = ((Form_pg_database) GETSTRUCT(tuple))->oid;
2924 : :
2925 : 520 : systable_endscan(scan);
2926 : :
2927 : : /*
2928 : : * Now that we have a database OID, we can try to lock the DB.
2929 : : */
2930 [ + - ]: 520 : if (lockmode != NoLock)
2931 : 520 : LockSharedObject(DatabaseRelationId, dbOid, 0, lockmode);
2932 : :
2933 : : /*
2934 : : * And now, re-fetch the tuple by OID. If it's still there and still
2935 : : * the same name, we win; else, drop the lock and loop back to try
2936 : : * again.
2937 : : */
2938 : 520 : tuple = SearchSysCache1(DATABASEOID, ObjectIdGetDatum(dbOid));
2939 [ + - ]: 520 : if (HeapTupleIsValid(tuple))
2940 : : {
2941 : 520 : Form_pg_database dbform = (Form_pg_database) GETSTRUCT(tuple);
2942 : :
2943 [ + - ]: 520 : if (strcmp(name, NameStr(dbform->datname)) == 0)
2944 : : {
2945 : : Datum datum;
2946 : : bool isnull;
2947 : :
2948 : : /* oid of the database */
2949 [ + - ]: 520 : if (dbIdP)
2950 : 520 : *dbIdP = dbOid;
2951 : : /* oid of the owner */
2952 [ + + ]: 520 : if (ownerIdP)
2953 : 440 : *ownerIdP = dbform->datdba;
2954 : : /* character encoding */
2955 [ + + ]: 520 : if (encodingP)
2956 : 440 : *encodingP = dbform->encoding;
2957 : : /* allowed as template? */
2958 [ + + ]: 520 : if (dbIsTemplateP)
2959 : 497 : *dbIsTemplateP = dbform->datistemplate;
2960 : : /* Has on login event trigger? */
2961 [ + + ]: 520 : if (dbHasLoginEvtP)
2962 : 440 : *dbHasLoginEvtP = dbform->dathasloginevt;
2963 : : /* allowing connections? */
2964 [ + + ]: 520 : if (dbAllowConnP)
2965 : 440 : *dbAllowConnP = dbform->datallowconn;
2966 : : /* limit of frozen XIDs */
2967 [ + + ]: 520 : if (dbFrozenXidP)
2968 : 440 : *dbFrozenXidP = dbform->datfrozenxid;
2969 : : /* minimum MultiXactId */
2970 [ + + ]: 520 : if (dbMinMultiP)
2971 : 440 : *dbMinMultiP = dbform->datminmxid;
2972 : : /* default tablespace for this database */
2973 [ + + ]: 520 : if (dbTablespace)
2974 : 454 : *dbTablespace = dbform->dattablespace;
2975 : : /* default locale settings for this database */
2976 [ + + ]: 520 : if (dbLocProvider)
2977 : 440 : *dbLocProvider = dbform->datlocprovider;
2978 [ + + ]: 520 : if (dbCollate)
2979 : : {
2980 : 440 : datum = SysCacheGetAttrNotNull(DATABASEOID, tuple, Anum_pg_database_datcollate);
2981 : 440 : *dbCollate = TextDatumGetCString(datum);
2982 : : }
2983 [ + + ]: 520 : if (dbCtype)
2984 : : {
2985 : 440 : datum = SysCacheGetAttrNotNull(DATABASEOID, tuple, Anum_pg_database_datctype);
2986 : 440 : *dbCtype = TextDatumGetCString(datum);
2987 : : }
2988 [ + + ]: 520 : if (dbLocale)
2989 : : {
2990 : 440 : datum = SysCacheGetAttr(DATABASEOID, tuple, Anum_pg_database_datlocale, &isnull);
2991 [ + + ]: 440 : if (isnull)
2992 : 411 : *dbLocale = NULL;
2993 : : else
2994 : 29 : *dbLocale = TextDatumGetCString(datum);
2995 : : }
2996 [ + + ]: 520 : if (dbIcurules)
2997 : : {
2998 : 440 : datum = SysCacheGetAttr(DATABASEOID, tuple, Anum_pg_database_daticurules, &isnull);
2999 [ + - ]: 440 : if (isnull)
3000 : 440 : *dbIcurules = NULL;
3001 : : else
3002 : 0 : *dbIcurules = TextDatumGetCString(datum);
3003 : : }
3004 [ + + ]: 520 : if (dbCollversion)
3005 : : {
3006 : 440 : datum = SysCacheGetAttr(DATABASEOID, tuple, Anum_pg_database_datcollversion, &isnull);
3007 [ + + ]: 440 : if (isnull)
3008 : 261 : *dbCollversion = NULL;
3009 : : else
3010 : 179 : *dbCollversion = TextDatumGetCString(datum);
3011 : : }
3012 : 520 : ReleaseSysCache(tuple);
3013 : 520 : result = true;
3014 : 520 : break;
3015 : : }
3016 : : /* can only get here if it was just renamed */
3017 : 0 : ReleaseSysCache(tuple);
3018 : : }
3019 : :
3020 [ # # ]: 0 : if (lockmode != NoLock)
3021 : 0 : UnlockSharedObject(DatabaseRelationId, dbOid, 0, lockmode);
3022 : : }
3023 : :
3024 : 541 : table_close(relation, AccessShareLock);
3025 : :
3026 : 541 : return result;
3027 : : }
3028 : :
3029 : : /* Check if current user has createdb privileges */
3030 : : bool
3031 : 491 : have_createdb_privilege(void)
3032 : : {
3033 : 491 : bool result = false;
3034 : : HeapTuple utup;
3035 : :
3036 : : /* Superusers can always do everything */
3037 [ + + ]: 491 : if (superuser())
3038 : 467 : return true;
3039 : :
3040 : 24 : utup = SearchSysCache1(AUTHOID, ObjectIdGetDatum(GetUserId()));
3041 [ + - ]: 24 : if (HeapTupleIsValid(utup))
3042 : : {
3043 : 24 : result = ((Form_pg_authid) GETSTRUCT(utup))->rolcreatedb;
3044 : 24 : ReleaseSysCache(utup);
3045 : : }
3046 : 24 : return result;
3047 : : }
3048 : :
3049 : : /*
3050 : : * Remove tablespace directories
3051 : : *
3052 : : * We don't know what tablespaces db_id is using, so iterate through all
3053 : : * tablespaces removing <tablespace>/db_id
3054 : : */
3055 : : static void
3056 : 58 : remove_dbtablespaces(Oid db_id)
3057 : : {
3058 : : Relation rel;
3059 : : TableScanDesc scan;
3060 : : HeapTuple tuple;
3061 : 58 : List *ltblspc = NIL;
3062 : : ListCell *cell;
3063 : : int ntblspc;
3064 : : int i;
3065 : : Oid *tablespace_ids;
3066 : :
3067 : 58 : rel = table_open(TableSpaceRelationId, AccessShareLock);
3068 : 58 : scan = table_beginscan_catalog(rel, 0, NULL);
3069 [ + + ]: 210 : while ((tuple = heap_getnext(scan, ForwardScanDirection)) != NULL)
3070 : : {
3071 : 152 : Form_pg_tablespace spcform = (Form_pg_tablespace) GETSTRUCT(tuple);
3072 : 152 : Oid dsttablespace = spcform->oid;
3073 : : char *dstpath;
3074 : : struct stat st;
3075 : :
3076 : : /* Don't mess with the global tablespace */
3077 [ + + ]: 152 : if (dsttablespace == GLOBALTABLESPACE_OID)
3078 : 94 : continue;
3079 : :
3080 : 94 : dstpath = GetDatabasePath(db_id, dsttablespace);
3081 : :
3082 [ + + - + ]: 94 : if (lstat(dstpath, &st) < 0 || !S_ISDIR(st.st_mode))
3083 : : {
3084 : : /* Assume we can ignore it */
3085 : 36 : pfree(dstpath);
3086 : 36 : continue;
3087 : : }
3088 : :
3089 [ - + ]: 58 : if (!rmtree(dstpath, true))
3090 [ # # ]: 0 : ereport(WARNING,
3091 : : (errmsg("some useless files may be left behind in old database directory \"%s\"",
3092 : : dstpath)));
3093 : :
3094 : 58 : ltblspc = lappend_oid(ltblspc, dsttablespace);
3095 : 58 : pfree(dstpath);
3096 : : }
3097 : :
3098 : 58 : ntblspc = list_length(ltblspc);
3099 [ + + ]: 58 : if (ntblspc == 0)
3100 : : {
3101 : 1 : table_endscan(scan);
3102 : 1 : table_close(rel, AccessShareLock);
3103 : 1 : return;
3104 : : }
3105 : :
3106 : 57 : tablespace_ids = (Oid *) palloc(ntblspc * sizeof(Oid));
3107 : 57 : i = 0;
3108 [ + - + + : 115 : foreach(cell, ltblspc)
+ + ]
3109 : 58 : tablespace_ids[i++] = lfirst_oid(cell);
3110 : :
3111 : : /* Record the filesystem change in XLOG */
3112 : : {
3113 : : xl_dbase_drop_rec xlrec;
3114 : :
3115 : 57 : xlrec.db_id = db_id;
3116 : 57 : xlrec.ntablespaces = ntblspc;
3117 : :
3118 : 57 : XLogBeginInsert();
3119 : 57 : XLogRegisterData(&xlrec, MinSizeOfDbaseDropRec);
3120 : 57 : XLogRegisterData(tablespace_ids, ntblspc * sizeof(Oid));
3121 : :
3122 : 57 : (void) XLogInsert(RM_DBASE_ID,
3123 : : XLOG_DBASE_DROP | XLR_SPECIAL_REL_UPDATE);
3124 : : }
3125 : :
3126 : 57 : list_free(ltblspc);
3127 : 57 : pfree(tablespace_ids);
3128 : :
3129 : 57 : table_endscan(scan);
3130 : 57 : table_close(rel, AccessShareLock);
3131 : : }
3132 : :
3133 : : /*
3134 : : * Check for existing files that conflict with a proposed new DB OID;
3135 : : * return true if there are any
3136 : : *
3137 : : * If there were a subdirectory in any tablespace matching the proposed new
3138 : : * OID, we'd get a create failure due to the duplicate name ... and then we'd
3139 : : * try to remove that already-existing subdirectory during the cleanup in
3140 : : * remove_dbtablespaces. Nuking existing files seems like a bad idea, so
3141 : : * instead we make this extra check before settling on the OID of the new
3142 : : * database. This exactly parallels what GetNewRelFileNumber() does for table
3143 : : * relfilenumber values.
3144 : : */
3145 : : static bool
3146 : 426 : check_db_file_conflict(Oid db_id)
3147 : : {
3148 : 426 : bool result = false;
3149 : : Relation rel;
3150 : : TableScanDesc scan;
3151 : : HeapTuple tuple;
3152 : :
3153 : 426 : rel = table_open(TableSpaceRelationId, AccessShareLock);
3154 : 426 : scan = table_beginscan_catalog(rel, 0, NULL);
3155 [ + + ]: 1349 : while ((tuple = heap_getnext(scan, ForwardScanDirection)) != NULL)
3156 : : {
3157 : 923 : Form_pg_tablespace spcform = (Form_pg_tablespace) GETSTRUCT(tuple);
3158 : 923 : Oid dsttablespace = spcform->oid;
3159 : : char *dstpath;
3160 : : struct stat st;
3161 : :
3162 : : /* Don't mess with the global tablespace */
3163 [ + + ]: 923 : if (dsttablespace == GLOBALTABLESPACE_OID)
3164 : 426 : continue;
3165 : :
3166 : 497 : dstpath = GetDatabasePath(db_id, dsttablespace);
3167 : :
3168 [ - + ]: 497 : if (lstat(dstpath, &st) == 0)
3169 : : {
3170 : : /* Found a conflicting file (or directory, whatever) */
3171 : 0 : pfree(dstpath);
3172 : 0 : result = true;
3173 : 0 : break;
3174 : : }
3175 : :
3176 : 497 : pfree(dstpath);
3177 : : }
3178 : :
3179 : 426 : table_endscan(scan);
3180 : 426 : table_close(rel, AccessShareLock);
3181 : :
3182 : 426 : return result;
3183 : : }
3184 : :
3185 : : /*
3186 : : * Issue a suitable errdetail message for a busy database
3187 : : */
3188 : : static int
3189 : 1 : errdetail_busy_db(int notherbackends, int npreparedxacts)
3190 : : {
3191 [ + - - + ]: 1 : if (notherbackends > 0 && npreparedxacts > 0)
3192 : :
3193 : : /*
3194 : : * We don't deal with singular versus plural here, since gettext
3195 : : * doesn't support multiple plurals in one string.
3196 : : */
3197 : 0 : errdetail("There are %d other session(s) and %d prepared transaction(s) using the database.",
3198 : : notherbackends, npreparedxacts);
3199 [ + - ]: 1 : else if (notherbackends > 0)
3200 : 1 : errdetail_plural("There is %d other session using the database.",
3201 : : "There are %d other sessions using the database.",
3202 : : notherbackends,
3203 : : notherbackends);
3204 : : else
3205 : 0 : errdetail_plural("There is %d prepared transaction using the database.",
3206 : : "There are %d prepared transactions using the database.",
3207 : : npreparedxacts,
3208 : : npreparedxacts);
3209 : 1 : return 0; /* just to keep ereport macro happy */
3210 : : }
3211 : :
3212 : : /*
3213 : : * get_database_oid - given a database name, look up the OID
3214 : : *
3215 : : * If missing_ok is false, throw an error if database name not found. If
3216 : : * true, just return InvalidOid.
3217 : : */
3218 : : Oid
3219 : 1731 : get_database_oid(const char *dbname, bool missing_ok)
3220 : : {
3221 : : Relation pg_database;
3222 : : ScanKeyData entry[1];
3223 : : SysScanDesc scan;
3224 : : HeapTuple dbtuple;
3225 : : Oid oid;
3226 : :
3227 : : /*
3228 : : * There's no syscache for pg_database indexed by name, so we must look
3229 : : * the hard way.
3230 : : */
3231 : 1731 : pg_database = table_open(DatabaseRelationId, AccessShareLock);
3232 : 1731 : ScanKeyInit(&entry[0],
3233 : : Anum_pg_database_datname,
3234 : : BTEqualStrategyNumber, F_NAMEEQ,
3235 : : CStringGetDatum(dbname));
3236 : 1731 : scan = systable_beginscan(pg_database, DatabaseNameIndexId, true,
3237 : : NULL, 1, entry);
3238 : :
3239 : 1731 : dbtuple = systable_getnext(scan);
3240 : :
3241 : : /* We assume that there can be at most one matching tuple */
3242 [ + + ]: 1731 : if (HeapTupleIsValid(dbtuple))
3243 : 1270 : oid = ((Form_pg_database) GETSTRUCT(dbtuple))->oid;
3244 : : else
3245 : 461 : oid = InvalidOid;
3246 : :
3247 : 1731 : systable_endscan(scan);
3248 : 1731 : table_close(pg_database, AccessShareLock);
3249 : :
3250 [ + + + + ]: 1731 : if (!OidIsValid(oid) && !missing_ok)
3251 [ + - ]: 4 : ereport(ERROR,
3252 : : (errcode(ERRCODE_UNDEFINED_DATABASE),
3253 : : errmsg("database \"%s\" does not exist",
3254 : : dbname)));
3255 : :
3256 : 1727 : return oid;
3257 : : }
3258 : :
3259 : :
3260 : : /*
3261 : : * While dropping a database the pg_database row is marked invalid, but the
3262 : : * catalog contents still exist. Connections to such a database are not
3263 : : * allowed.
3264 : : */
3265 : : bool
3266 : 29895 : database_is_invalid_form(Form_pg_database datform)
3267 : : {
3268 : 29895 : return datform->datconnlimit == DATCONNLIMIT_INVALID_DB;
3269 : : }
3270 : :
3271 : :
3272 : : /*
3273 : : * Convenience wrapper around database_is_invalid_form()
3274 : : */
3275 : : bool
3276 : 440 : database_is_invalid_oid(Oid dboid)
3277 : : {
3278 : : HeapTuple dbtup;
3279 : : Form_pg_database dbform;
3280 : : bool invalid;
3281 : :
3282 : 440 : dbtup = SearchSysCache1(DATABASEOID, ObjectIdGetDatum(dboid));
3283 [ - + ]: 440 : if (!HeapTupleIsValid(dbtup))
3284 [ # # ]: 0 : elog(ERROR, "cache lookup failed for database %u", dboid);
3285 : 440 : dbform = (Form_pg_database) GETSTRUCT(dbtup);
3286 : :
3287 : 440 : invalid = database_is_invalid_form(dbform);
3288 : :
3289 : 440 : ReleaseSysCache(dbtup);
3290 : :
3291 : 440 : return invalid;
3292 : : }
3293 : :
3294 : :
3295 : : /*
3296 : : * recovery_create_dbdir()
3297 : : *
3298 : : * During recovery, there's a case where we validly need to recover a missing
3299 : : * tablespace directory so that recovery can continue. This happens when
3300 : : * recovery wants to create a database but the holding tablespace has been
3301 : : * removed before the server stopped. Since we expect that the directory will
3302 : : * be gone before reaching recovery consistency, and we have no knowledge about
3303 : : * the tablespace other than its OID here, we create a real directory under
3304 : : * pg_tblspc here instead of restoring the symlink.
3305 : : *
3306 : : * If only_tblspc is true, then the requested directory must be in pg_tblspc/
3307 : : */
3308 : : static void
3309 : 24 : recovery_create_dbdir(char *path, bool only_tblspc)
3310 : : {
3311 : : struct stat st;
3312 : :
3313 : : Assert(RecoveryInProgress());
3314 : :
3315 [ + - ]: 24 : if (stat(path, &st) == 0)
3316 : 24 : return;
3317 : :
3318 [ # # # # ]: 0 : if (only_tblspc && strstr(path, PG_TBLSPC_DIR_SLASH) == NULL)
3319 [ # # ]: 0 : elog(PANIC, "requested to created invalid directory: %s", path);
3320 : :
3321 [ # # # # ]: 0 : if (reachedConsistency && !allow_in_place_tablespaces)
3322 [ # # ]: 0 : ereport(PANIC,
3323 : : errmsg("missing directory \"%s\"", path));
3324 : :
3325 [ # # # # ]: 0 : elog(reachedConsistency ? WARNING : DEBUG1,
3326 : : "creating missing directory: %s", path);
3327 : :
3328 [ # # ]: 0 : if (pg_mkdir_p(path, pg_dir_create_mode) != 0)
3329 [ # # ]: 0 : ereport(PANIC,
3330 : : errmsg("could not create missing directory \"%s\": %m", path));
3331 : : }
3332 : :
3333 : :
3334 : : /*
3335 : : * DATABASE resource manager's routines
3336 : : */
3337 : : void
3338 : 44 : dbase_redo(XLogReaderState *record)
3339 : : {
3340 : 44 : uint8 info = XLogRecGetInfo(record) & ~XLR_INFO_MASK;
3341 : :
3342 : : /* Backup blocks are not used in dbase records */
3343 : : Assert(!XLogRecHasAnyBlockRefs(record));
3344 : :
3345 [ + + ]: 44 : if (info == XLOG_DBASE_CREATE_FILE_COPY)
3346 : : {
3347 : 5 : xl_dbase_create_file_copy_rec *xlrec =
3348 : 5 : (xl_dbase_create_file_copy_rec *) XLogRecGetData(record);
3349 : : char *src_path;
3350 : : char *dst_path;
3351 : : char *parent_path;
3352 : : struct stat st;
3353 : :
3354 : 5 : src_path = GetDatabasePath(xlrec->src_db_id, xlrec->src_tablespace_id);
3355 : 5 : dst_path = GetDatabasePath(xlrec->db_id, xlrec->tablespace_id);
3356 : :
3357 : : /*
3358 : : * Our theory for replaying a CREATE is to forcibly drop the target
3359 : : * subdirectory if present, then re-copy the source data. This may be
3360 : : * more work than needed, but it is simple to implement.
3361 : : */
3362 [ - + - - ]: 5 : if (stat(dst_path, &st) == 0 && S_ISDIR(st.st_mode))
3363 : : {
3364 [ # # ]: 0 : if (!rmtree(dst_path, true))
3365 : : /* If this failed, copydir() below is going to error. */
3366 [ # # ]: 0 : ereport(WARNING,
3367 : : (errmsg("some useless files may be left behind in old database directory \"%s\"",
3368 : : dst_path)));
3369 : : }
3370 : :
3371 : : /*
3372 : : * If the parent of the target path doesn't exist, create it now. This
3373 : : * enables us to create the target underneath later.
3374 : : */
3375 : 5 : parent_path = pstrdup(dst_path);
3376 : 5 : get_parent_directory(parent_path);
3377 [ - + ]: 5 : if (stat(parent_path, &st) < 0)
3378 : : {
3379 [ # # ]: 0 : if (errno != ENOENT)
3380 [ # # ]: 0 : ereport(FATAL,
3381 : : errmsg("could not stat directory \"%s\": %m",
3382 : : dst_path));
3383 : :
3384 : : /* create the parent directory if needed and valid */
3385 : 0 : recovery_create_dbdir(parent_path, true);
3386 : : }
3387 : 5 : pfree(parent_path);
3388 : :
3389 : : /*
3390 : : * There's a case where the copy source directory is missing for the
3391 : : * same reason above. Create the empty source directory so that
3392 : : * copydir below doesn't fail. The directory will be dropped soon by
3393 : : * recovery.
3394 : : */
3395 [ - + - - ]: 5 : if (stat(src_path, &st) < 0 && errno == ENOENT)
3396 : 0 : recovery_create_dbdir(src_path, false);
3397 : :
3398 : : /*
3399 : : * Force dirty buffers out to disk, to ensure source database is
3400 : : * up-to-date for the copy.
3401 : : */
3402 : 5 : FlushDatabaseBuffers(xlrec->src_db_id);
3403 : :
3404 : : /* Close all smgr fds in all backends. */
3405 : 5 : WaitForProcSignalBarrier(EmitProcSignalBarrier(PROCSIGNAL_BARRIER_SMGRRELEASE));
3406 : :
3407 : : /*
3408 : : * Copy this subdirectory to the new location
3409 : : *
3410 : : * We don't need to copy subdirectories
3411 : : */
3412 : 5 : copydir(src_path, dst_path, false);
3413 : :
3414 : 5 : pfree(src_path);
3415 : 5 : pfree(dst_path);
3416 : : }
3417 [ + + ]: 39 : else if (info == XLOG_DBASE_CREATE_WAL_LOG)
3418 : : {
3419 : 24 : xl_dbase_create_wal_log_rec *xlrec =
3420 : 24 : (xl_dbase_create_wal_log_rec *) XLogRecGetData(record);
3421 : : char *dbpath;
3422 : : char *parent_path;
3423 : :
3424 : 24 : dbpath = GetDatabasePath(xlrec->db_id, xlrec->tablespace_id);
3425 : :
3426 : : /* create the parent directory if needed and valid */
3427 : 24 : parent_path = pstrdup(dbpath);
3428 : 24 : get_parent_directory(parent_path);
3429 : 24 : recovery_create_dbdir(parent_path, true);
3430 : 24 : pfree(parent_path);
3431 : :
3432 : : /* Create the database directory with the version file. */
3433 : 24 : CreateDirAndVersionFile(dbpath, xlrec->db_id, xlrec->tablespace_id,
3434 : : true);
3435 : 24 : pfree(dbpath);
3436 : : }
3437 [ + - ]: 15 : else if (info == XLOG_DBASE_DROP)
3438 : : {
3439 : 15 : xl_dbase_drop_rec *xlrec = (xl_dbase_drop_rec *) XLogRecGetData(record);
3440 : : char *dst_path;
3441 : : int i;
3442 : :
3443 [ + - ]: 15 : if (InHotStandby)
3444 : : {
3445 : : /*
3446 : : * Lock database while we resolve conflicts to ensure that
3447 : : * InitPostgres() cannot fully re-execute concurrently. This
3448 : : * avoids backends re-connecting automatically to same database,
3449 : : * which can happen in some cases.
3450 : : *
3451 : : * This will lock out walsenders trying to connect to db-specific
3452 : : * slots for logical decoding too, so it's safe for us to drop
3453 : : * slots.
3454 : : */
3455 : 15 : LockSharedObjectForSession(DatabaseRelationId, xlrec->db_id, 0, AccessExclusiveLock);
3456 : 15 : ResolveRecoveryConflictWithDatabase(xlrec->db_id);
3457 : : }
3458 : :
3459 : : /* Drop any database-specific replication slots */
3460 : 15 : ReplicationSlotsDropDBSlots(xlrec->db_id);
3461 : :
3462 : : /* Drop pages for this database that are in the shared buffer cache */
3463 : 15 : DropDatabaseBuffers(xlrec->db_id);
3464 : :
3465 : : /* Also, clean out any fsync requests that might be pending in md.c */
3466 : 15 : ForgetDatabaseSyncRequests(xlrec->db_id);
3467 : :
3468 : : /* Clean out the xlog relcache too */
3469 : 15 : XLogDropDatabase(xlrec->db_id);
3470 : :
3471 : : /* Close all smgr fds in all backends. */
3472 : 15 : WaitForProcSignalBarrier(EmitProcSignalBarrier(PROCSIGNAL_BARRIER_SMGRRELEASE));
3473 : :
3474 [ + + ]: 30 : for (i = 0; i < xlrec->ntablespaces; i++)
3475 : : {
3476 : 15 : dst_path = GetDatabasePath(xlrec->db_id, xlrec->tablespace_ids[i]);
3477 : :
3478 : : /* And remove the physical files */
3479 [ + + ]: 15 : if (!rmtree(dst_path, true))
3480 [ + - ]: 1 : ereport(WARNING,
3481 : : (errmsg("some useless files may be left behind in old database directory \"%s\"",
3482 : : dst_path)));
3483 : 15 : pfree(dst_path);
3484 : : }
3485 : :
3486 [ + - ]: 15 : if (InHotStandby)
3487 : : {
3488 : : /*
3489 : : * Release locks prior to commit. XXX There is a race condition
3490 : : * here that may allow backends to reconnect, but the window for
3491 : : * this is small because the gap between here and commit is mostly
3492 : : * fairly small and it is unlikely that people will be dropping
3493 : : * databases that we are trying to connect to anyway.
3494 : : */
3495 : 15 : UnlockSharedObjectForSession(DatabaseRelationId, xlrec->db_id, 0, AccessExclusiveLock);
3496 : : }
3497 : : }
3498 : : else
3499 [ # # ]: 0 : elog(PANIC, "dbase_redo: unknown op code %u", info);
3500 : 44 : }
|
