LCOV - code coverage report
Current view: top level - src/backend/commands - dbcommands.c (source / functions) Hit Total Coverage
Test: PostgreSQL 16beta1 Lines: 825 1031 80.0 %
Date: 2023-05-30 16:15:03 Functions: 24 28 85.7 %
Legend: Lines: hit not hit

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

Generated by: LCOV version 1.14