Quelle  dbcommands.c

/*-------------------------------------------------------------------------
 *
 * dbcommands.c
 *  Database management commands (create/drop database).
 *
 * Note: database creation/destruction commands use exclusive locks on
 * the database objects (as expressed by LockSharedObject()) to avoid
 * stepping on each others' toes.  Formerly we used table-level locks
 * on pg_database, but that's too coarse-grained.
 *
 * Portions Copyright (c) 1996-2025, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 *
 * IDENTIFICATION
 *   src/backend/commands/dbcommands.c
 *
 *-------------------------------------------------------------------------
 */
#include "postgres.h"

#include <fcntl.h>
#include <unistd.h>
#include <sys/stat.h>

#include "access/genam.h"
#include "access/heapam.h"
#include "access/htup_details.h"
#include "access/multixact.h"
#include "access/tableam.h"
#include "access/xact.h"
#include "access/xloginsert.h"
#include "access/xlogrecovery.h"
#include "access/xlogutils.h"
#include "catalog/catalog.h"
#include "catalog/dependency.h"
#include "catalog/indexing.h"
#include "catalog/objectaccess.h"
#include "catalog/pg_authid.h"
#include "catalog/pg_collation.h"
#include "catalog/pg_database.h"
#include "catalog/pg_db_role_setting.h"
#include "catalog/pg_subscription.h"
#include "catalog/pg_tablespace.h"
#include "commands/comment.h"
#include "commands/dbcommands.h"
#include "commands/dbcommands_xlog.h"
#include "commands/defrem.h"
#include "commands/seclabel.h"
#include "commands/tablespace.h"
#include "common/file_perm.h"
#include "mb/pg_wchar.h"
#include "miscadmin.h"
#include "pgstat.h"
#include "postmaster/bgwriter.h"
#include "replication/slot.h"
#include "storage/copydir.h"
#include "storage/fd.h"
#include "storage/ipc.h"
#include "storage/lmgr.h"
#include "storage/md.h"
#include "storage/procarray.h"
#include "storage/smgr.h"
#include "utils/acl.h"
#include "utils/builtins.h"
#include "utils/fmgroids.h"
#include "utils/pg_locale.h"
#include "utils/relmapper.h"
#include "utils/snapmgr.h"
#include "utils/syscache.h"

/*
 * Create database strategy.
 *
 * CREATEDB_WAL_LOG will copy the database at the block level and WAL log each
 * copied block.
 *
 * CREATEDB_FILE_COPY will simply perform a file system level copy of the
 * database and log a single record for each tablespace copied. To make this
 * safe, it also triggers checkpoints before and after the operation.
 */
typedef enum CreateDBStrategy
{
 CREATEDB_WAL_LOG,
 CREATEDB_FILE_COPY,
} CreateDBStrategy;

typedef struct
{
 Oid   src_dboid;  /* source (template) DB */
 Oid   dest_dboid;  /* DB we are trying to create */
 CreateDBStrategy strategy; /* create db strategy */
} createdb_failure_params;

typedef struct
{
 Oid   dest_dboid;  /* DB we are trying to move */
 Oid   dest_tsoid;  /* tablespace we are trying to move to */
} movedb_failure_params;

/*
 * Information about a relation to be copied when creating a database.
 */
typedef struct CreateDBRelInfo
{
 RelFileLocator rlocator; /* physical relation identifier */
 Oid   reloid;   /* relation oid */
 bool  permanent;  /* relation is permanent or unlogged */
} CreateDBRelInfo;


/* non-export function prototypes */
static void createdb_failure_callback(int code, Datum arg);
static void movedb(const char *dbname, const char *tblspcname);
static void movedb_failure_callback(int code, Datum arg);
static bool get_db_info(const char *name, LOCKMODE lockmode,
      Oid *dbIdP, Oid *ownerIdP,
      int *encodingP, bool *dbIsTemplateP, bool *dbAllowConnP, bool *dbHasLoginEvtP,
      TransactionId *dbFrozenXidP, MultiXactId *dbMinMultiP,
      Oid *dbTablespace, char **dbCollate, char **dbCtype, char **dbLocale,
      char **dbIcurules,
      char *dbLocProvider,
      char **dbCollversion);
static void remove_dbtablespaces(Oid db_id);
static bool check_db_file_conflict(Oid db_id);
static int errdetail_busy_db(int notherbackends, int npreparedxacts);
static void CreateDatabaseUsingWalLog(Oid src_dboid, Oid dst_dboid, Oid src_tsid,
           Oid dst_tsid);
static List *ScanSourceDatabasePgClass(Oid tbid, Oid dbid, char *srcpath);
static List *ScanSourceDatabasePgClassPage(Page page, Buffer buf, Oid tbid,
             Oid dbid, char *srcpath,
             List *rlocatorlist, Snapshot snapshot);
static CreateDBRelInfo *ScanSourceDatabasePgClassTuple(HeapTupleData *tuple,
                Oid tbid, Oid dbid,
                char *srcpath);
static void CreateDirAndVersionFile(char *dbpath, Oid dbid, Oid tsid,
         bool isRedo);
static void CreateDatabaseUsingFileCopy(Oid src_dboid, Oid dst_dboid,
          Oid src_tsid, Oid dst_tsid);
static void recovery_create_dbdir(char *path, bool only_tblspc);

/*
 * Create a new database using the WAL_LOG strategy.
 *
 * Each copied block is separately written to the write-ahead log.
 */
static void
CreateDatabaseUsingWalLog(Oid src_dboid, Oid dst_dboid,
        Oid src_tsid, Oid dst_tsid)
{
 char    *srcpath;
 char    *dstpath;
 List    *rlocatorlist = NULL;
 ListCell   *cell;
 LockRelId srcrelid;
 LockRelId dstrelid;
 RelFileLocator srcrlocator;
 RelFileLocator dstrlocator;
 CreateDBRelInfo *relinfo;

 /* Get source and destination database paths. */
 srcpath = GetDatabasePath(src_dboid, src_tsid);
 dstpath = GetDatabasePath(dst_dboid, dst_tsid);

 /* Create database directory and write PG_VERSION file. */
 CreateDirAndVersionFile(dstpath, dst_dboid, dst_tsid, false);

 /* Copy relmap file from source database to the destination database. */
 RelationMapCopy(dst_dboid, dst_tsid, srcpath, dstpath);

 /* Get list of relfilelocators to copy from the source database. */
 rlocatorlist = ScanSourceDatabasePgClass(src_tsid, src_dboid, srcpath);
 Assert(rlocatorlist != NIL);

 /*
  * Database IDs will be the same for all relations so set them before
  * entering the loop.
 */
 srcrelid.dbId = src_dboid;
 dstrelid.dbId = dst_dboid;

 /* Loop over our list of relfilelocators and copy each one. */
 foreach(cell, rlocatorlist)
 {
  relinfo = lfirst(cell);
  srcrlocator = relinfo->rlocator;

  /*
   * If the relation is from the source db's default tablespace then we
   * need to create it in the destination db's default tablespace.
   * Otherwise, we need to create in the same tablespace as it is in the
   * source database.
 */
  if (srcrlocator.spcOid == src_tsid)
   dstrlocator.spcOid = dst_tsid;
  else
   dstrlocator.spcOid = srcrlocator.spcOid;

  dstrlocator.dbOid = dst_dboid;
  dstrlocator.relNumber = srcrlocator.relNumber;

  /*
   * Acquire locks on source and target relations before copying.
   *
   * We typically do not read relation data into shared_buffers without
   * holding a relation lock. It's unclear what could go wrong if we
   * skipped it in this case, because nobody can be modifying either the
   * source or destination database at this point, and we have locks on
   * both databases, too, but let's take the conservative route.
 */
  dstrelid.relId = srcrelid.relId = relinfo->reloid;
  LockRelationId(&srcrelid, AccessShareLock);
  LockRelationId(&dstrelid, AccessShareLock);

  /* Copy relation storage from source to the destination. */
  CreateAndCopyRelationData(srcrlocator, dstrlocator, relinfo->permanent);

  /* Release the relation locks. */
  UnlockRelationId(&srcrelid, AccessShareLock);
  UnlockRelationId(&dstrelid, AccessShareLock);
 }

 pfree(srcpath);
 pfree(dstpath);
 list_free_deep(rlocatorlist);
}

/*
 * Scan the pg_class table in the source database to identify the relations
 * that need to be copied to the destination database.
 *
 * This is an exception to the usual rule that cross-database access is
 * not possible. We can make it work here because we know that there are no
 * connections to the source database and (since there can't be prepared
 * transactions touching that database) no in-doubt tuples either. This
 * means that we don't need to worry about pruning removing anything from
 * under us, and we don't need to be too picky about our snapshot either.
 * As long as it sees all previously-committed XIDs as committed and all
 * aborted XIDs as aborted, we should be fine: nothing else is possible
 * here.
 *
 * We can't rely on the relcache for anything here, because that only knows
 * about the database to which we are connected, and can't handle access to
 * other databases. That also means we can't rely on the heap scan
 * infrastructure, which would be a bad idea anyway since it might try
 * to do things like HOT pruning which we definitely can't do safely in
 * a database to which we're not even connected.
 */
static List *
ScanSourceDatabasePgClass(Oid tbid, Oid dbid, char *srcpath)
{
 RelFileLocator rlocator;
 BlockNumber nblocks;
 BlockNumber blkno;
 Buffer  buf;
 RelFileNumber relfilenumber;
 Page  page;
 List    *rlocatorlist = NIL;
 LockRelId relid;
 Snapshot snapshot;
 SMgrRelation smgr;
 BufferAccessStrategy bstrategy;

 /* Get pg_class relfilenumber. */
 relfilenumber = RelationMapOidToFilenumberForDatabase(srcpath,
                RelationRelationId);

 /* Don't read data into shared_buffers without holding a relation lock. */
 relid.dbId = dbid;
 relid.relId = RelationRelationId;
 LockRelationId(&relid, AccessShareLock);

 /* Prepare a RelFileLocator for the pg_class relation. */
 rlocator.spcOid = tbid;
 rlocator.dbOid = dbid;
 rlocator.relNumber = relfilenumber;

 smgr = smgropen(rlocator, INVALID_PROC_NUMBER);
 nblocks = smgrnblocks(smgr, MAIN_FORKNUM);
 smgrclose(smgr);

 /* Use a buffer access strategy since this is a bulk read operation. */
 bstrategy = GetAccessStrategy(BAS_BULKREAD);

 /*
  * As explained in the function header comments, we need a snapshot that
  * will see all committed transactions as committed, and our transaction
  * snapshot - or the active snapshot - might not be new enough for that,
  * but the return value of GetLatestSnapshot() should work fine.
 */
 snapshot = RegisterSnapshot(GetLatestSnapshot());

 /* Process the relation block by block. */
 for (blkno = 0; blkno < nblocks; blkno++)
 {
  CHECK_FOR_INTERRUPTS();

  buf = ReadBufferWithoutRelcache(rlocator, MAIN_FORKNUM, blkno,
          RBM_NORMAL, bstrategy, true);

  LockBuffer(buf, BUFFER_LOCK_SHARE);
  page = BufferGetPage(buf);
  if (PageIsNew(page) || PageIsEmpty(page))
  {
   UnlockReleaseBuffer(buf);
   continue;
  }

  /* Append relevant pg_class tuples for current page to rlocatorlist. */
  rlocatorlist = ScanSourceDatabasePgClassPage(page, buf, tbid, dbid,
              srcpath, rlocatorlist,
              snapshot);

  UnlockReleaseBuffer(buf);
 }
 UnregisterSnapshot(snapshot);

 /* Release relation lock. */
 UnlockRelationId(&relid, AccessShareLock);

 return rlocatorlist;
}

/*
 * Scan one page of the source database's pg_class relation and add relevant
 * entries to rlocatorlist. The return value is the updated list.
 */
static List *
ScanSourceDatabasePgClassPage(Page page, Buffer buf, Oid tbid, Oid dbid,
         char *srcpath, List *rlocatorlist,
         Snapshot snapshot)
{
 BlockNumber blkno = BufferGetBlockNumber(buf);
 OffsetNumber offnum;
 OffsetNumber maxoff;
 HeapTupleData tuple;

 maxoff = PageGetMaxOffsetNumber(page);

 /* Loop over offsets. */
 for (offnum = FirstOffsetNumber;
   offnum <= maxoff;
   offnum = OffsetNumberNext(offnum))
 {
  ItemId  itemid;

  itemid = PageGetItemId(page, offnum);

  /* Nothing to do if slot is empty or already dead. */
  if (!ItemIdIsUsed(itemid) || ItemIdIsDead(itemid) ||
   ItemIdIsRedirected(itemid))
   continue;

  Assert(ItemIdIsNormal(itemid));
  ItemPointerSet(&(tuple.t_self), blkno, offnum);

  /* Initialize a HeapTupleData structure. */
  tuple.t_data = (HeapTupleHeader) PageGetItem(page, itemid);
  tuple.t_len = ItemIdGetLength(itemid);
  tuple.t_tableOid = RelationRelationId;

  /* Skip tuples that are not visible to this snapshot. */
  if (HeapTupleSatisfiesVisibility(&tuple, snapshot, buf))
  {
   CreateDBRelInfo *relinfo;

   /*
    * ScanSourceDatabasePgClassTuple is in charge of constructing a
    * CreateDBRelInfo object for this tuple, but can also decide that
    * this tuple isn't something we need to copy. If we do need to
    * copy the relation, add it to the list.
 */
   relinfo = ScanSourceDatabasePgClassTuple(&tuple, tbid, dbid,
              srcpath);
   if (relinfo != NULL)
    rlocatorlist = lappend(rlocatorlist, relinfo);
  }
 }

 return rlocatorlist;
}

/*
 * Decide whether a certain pg_class tuple represents something that
 * needs to be copied from the source database to the destination database,
 * and if so, construct a CreateDBRelInfo for it.
 *
 * Visibility checks are handled by the caller, so our job here is just
 * to assess the data stored in the tuple.
 */
CreateDBRelInfo *
ScanSourceDatabasePgClassTuple(HeapTupleData *tuple, Oid tbid, Oid dbid,
          char *srcpath)
{
 CreateDBRelInfo *relinfo;
 Form_pg_class classForm;
 RelFileNumber relfilenumber = InvalidRelFileNumber;

 classForm = (Form_pg_class) GETSTRUCT(tuple);

 /*
  * Return NULL if this object does not need to be copied.
  *
  * Shared objects don't need to be copied, because they are shared.
  * Objects without storage can't be copied, because there's nothing to
  * copy. Temporary relations don't need to be copied either, because they
  * are inaccessible outside of the session that created them, which must
  * be gone already, and couldn't connect to a different database if it
  * still existed. autovacuum will eventually remove the pg_class entries
  * as well.
 */
 if (classForm->reltablespace == GLOBALTABLESPACE_OID ||
  !RELKIND_HAS_STORAGE(classForm->relkind) ||
  classForm->relpersistence == RELPERSISTENCE_TEMP)
  return NULL;

 /*
  * If relfilenumber is valid then directly use it.  Otherwise, consult the
  * relmap.
 */
 if (RelFileNumberIsValid(classForm->relfilenode))
  relfilenumber = classForm->relfilenode;
 else
  relfilenumber = RelationMapOidToFilenumberForDatabase(srcpath,
                 classForm->oid);

 /* We must have a valid relfilenumber. */
 if (!RelFileNumberIsValid(relfilenumber))
  elog(ERROR, "relation with OID %u does not have a valid relfilenumber",
    classForm->oid);

 /* Prepare a rel info element and add it to the list. */
 relinfo = (CreateDBRelInfo *) palloc(sizeof(CreateDBRelInfo));
 if (OidIsValid(classForm->reltablespace))
  relinfo->rlocator.spcOid = classForm->reltablespace;
 else
  relinfo->rlocator.spcOid = tbid;

 relinfo->rlocator.dbOid = dbid;
 relinfo->rlocator.relNumber = relfilenumber;
 relinfo->reloid = classForm->oid;

 /* Temporary relations were rejected above. */
 Assert(classForm->relpersistence != RELPERSISTENCE_TEMP);
 relinfo->permanent =
  (classForm->relpersistence == RELPERSISTENCE_PERMANENT) ? true : false;

 return relinfo;
}

/*
 * Create database directory and write out the PG_VERSION file in the database
 * path.  If isRedo is true, it's okay for the database directory to exist
 * already.
 */
static void
CreateDirAndVersionFile(char *dbpath, Oid dbid, Oid tsid, bool isRedo)
{
 int   fd;
 int   nbytes;
 char  versionfile[MAXPGPATH];
 char  buf[16];

 /*
  * Note that we don't have to copy version data from the source database;
  * there's only one legal value.
 */
 sprintf(buf, "%s\n", PG_MAJORVERSION);
 nbytes = strlen(PG_MAJORVERSION) + 1;

 /* Create database directory. */
 if (MakePGDirectory(dbpath) < 0)
 {
  /* Failure other than already exists or not in WAL replay? */
  if (errno != EEXIST || !isRedo)
   ereport(ERROR,
     (errcode_for_file_access(),
      errmsg("could not create directory \"%s\": %m", dbpath)));
 }

 /*
  * Create PG_VERSION file in the database path.  If the file already
  * exists and we are in WAL replay then try again to open it in write
  * mode.
 */
 snprintf(versionfile, sizeof(versionfile), "%s/%s", dbpath, "PG_VERSION");

 fd = OpenTransientFile(versionfile, O_WRONLY | O_CREAT | O_EXCL | PG_BINARY);
 if (fd < 0 && errno == EEXIST && isRedo)
  fd = OpenTransientFile(versionfile, O_WRONLY | O_TRUNC | PG_BINARY);

 if (fd < 0)
  ereport(ERROR,
    (errcode_for_file_access(),
     errmsg("could not create file \"%s\": %m", versionfile)));

 /* Write PG_MAJORVERSION in the PG_VERSION file. */
 pgstat_report_wait_start(WAIT_EVENT_VERSION_FILE_WRITE);
 errno = 0;
 if ((int) write(fd, buf, nbytes) != nbytes)
 {
  /* If write didn't set errno, assume problem is no disk space. */
  if (errno == 0)
   errno = ENOSPC;
  ereport(ERROR,
    (errcode_for_file_access(),
     errmsg("could not write to file \"%s\": %m", versionfile)));
 }
 pgstat_report_wait_end();

 pgstat_report_wait_start(WAIT_EVENT_VERSION_FILE_SYNC);
 if (pg_fsync(fd) != 0)
  ereport(data_sync_elevel(ERROR),
    (errcode_for_file_access(),
     errmsg("could not fsync file \"%s\": %m", versionfile)));
 fsync_fname(dbpath, true);
 pgstat_report_wait_end();

 /* Close the version file. */
 CloseTransientFile(fd);

 /* If we are not in WAL replay then write the WAL. */
 if (!isRedo)
 {
  xl_dbase_create_wal_log_rec xlrec;

  START_CRIT_SECTION();

  xlrec.db_id = dbid;
  xlrec.tablespace_id = tsid;

  XLogBeginInsert();
  XLogRegisterData(&xlrec,
       sizeof(xl_dbase_create_wal_log_rec));

  (void) XLogInsert(RM_DBASE_ID, XLOG_DBASE_CREATE_WAL_LOG);

  END_CRIT_SECTION();
 }
}

/*
 * Create a new database using the FILE_COPY strategy.
 *
 * Copy each tablespace at the filesystem level, and log a single WAL record
 * for each tablespace copied.  This requires a checkpoint before and after the
 * copy, which may be expensive, but it does greatly reduce WAL generation
 * if the copied database is large.
 */
static void
CreateDatabaseUsingFileCopy(Oid src_dboid, Oid dst_dboid, Oid src_tsid,
       Oid dst_tsid)
{
 TableScanDesc scan;
 Relation rel;
 HeapTuple tuple;

 /*
  * Force a checkpoint before starting the copy. This will force all dirty
  * buffers, including those of unlogged tables, out to disk, to ensure
  * source database is up-to-date on disk for the copy.
  * FlushDatabaseBuffers() would suffice for that, but we also want to
  * process any pending unlink requests. Otherwise, if a checkpoint
  * happened while we're copying files, a file might be deleted just when
  * we're about to copy it, causing the lstat() call in copydir() to fail
  * with ENOENT.
  *
  * In binary upgrade mode, we can skip this checkpoint because pg_upgrade
  * is careful to ensure that template0 is fully written to disk prior to
  * any CREATE DATABASE commands.
 */
 if (!IsBinaryUpgrade)
  RequestCheckpoint(CHECKPOINT_IMMEDIATE | CHECKPOINT_FORCE |
        CHECKPOINT_WAIT | CHECKPOINT_FLUSH_ALL);

 /*
  * Iterate through all tablespaces of the template database, and copy each
  * one to the new database.
 */
 rel = table_open(TableSpaceRelationId, AccessShareLock);
 scan = table_beginscan_catalog(rel, 0, NULL);
 while ((tuple = heap_getnext(scan, ForwardScanDirection)) != NULL)
 {
  Form_pg_tablespace spaceform = (Form_pg_tablespace) GETSTRUCT(tuple);
  Oid   srctablespace = spaceform->oid;
  Oid   dsttablespace;
  char    *srcpath;
  char    *dstpath;
  struct stat st;

  /* No need to copy global tablespace */
  if (srctablespace == GLOBALTABLESPACE_OID)
   continue;

  srcpath = GetDatabasePath(src_dboid, srctablespace);

  if (stat(srcpath, &st) < 0 || !S_ISDIR(st.st_mode) ||
   directory_is_empty(srcpath))
  {
   /* Assume we can ignore it */
   pfree(srcpath);
   continue;
  }

  if (srctablespace == src_tsid)
   dsttablespace = dst_tsid;
  else
   dsttablespace = srctablespace;

  dstpath = GetDatabasePath(dst_dboid, dsttablespace);

  /*
   * Copy this subdirectory to the new location
   *
   * We don't need to copy subdirectories
 */
  copydir(srcpath, dstpath, false);

  /* Record the filesystem change in XLOG */
  {
   xl_dbase_create_file_copy_rec xlrec;

   xlrec.db_id = dst_dboid;
   xlrec.tablespace_id = dsttablespace;
   xlrec.src_db_id = src_dboid;
   xlrec.src_tablespace_id = srctablespace;

   XLogBeginInsert();
   XLogRegisterData(&xlrec,
        sizeof(xl_dbase_create_file_copy_rec));

   (void) XLogInsert(RM_DBASE_ID,
         XLOG_DBASE_CREATE_FILE_COPY | XLR_SPECIAL_REL_UPDATE);
  }
  pfree(srcpath);
  pfree(dstpath);
 }
 table_endscan(scan);
 table_close(rel, AccessShareLock);

 /*
  * We force a checkpoint before committing.  This effectively means that
  * committed XLOG_DBASE_CREATE_FILE_COPY operations will never need to be
  * replayed (at least not in ordinary crash recovery; we still have to
  * make the XLOG entry for the benefit of PITR operations). This avoids
  * two nasty scenarios:
  *
  * #1: At wal_level=minimal, we don't XLOG the contents of newly created
  * relfilenodes; therefore the drop-and-recreate-whole-directory behavior
  * of DBASE_CREATE replay would lose such files created in the new
  * database between our commit and the next checkpoint.
  *
  * #2: Since we have to recopy the source database during DBASE_CREATE
  * replay, we run the risk of copying changes in it that were committed
  * after the original CREATE DATABASE command but before the system crash
  * that led to the replay.  This is at least unexpected and at worst could
  * lead to inconsistencies, eg duplicate table names.
  *
  * (Both of these were real bugs in releases 8.0 through 8.0.3.)
  *
  * In PITR replay, the first of these isn't an issue, and the second is
  * only a risk if the CREATE DATABASE and subsequent template database
  * change both occur while a base backup is being taken. There doesn't
  * seem to be much we can do about that except document it as a
  * limitation.
  *
  * In binary upgrade mode, we can skip this checkpoint because neither of
  * these problems applies: we don't ever replay the WAL generated during
  * pg_upgrade, and we don't support taking base backups during pg_upgrade
  * (not to mention that we don't concurrently modify template0, either).
  *
  * See CreateDatabaseUsingWalLog() for a less cheesy CREATE DATABASE
  * strategy that avoids these problems.
 */
 if (!IsBinaryUpgrade)
  RequestCheckpoint(CHECKPOINT_IMMEDIATE | CHECKPOINT_FORCE |
        CHECKPOINT_WAIT);
}

/*
 * CREATE DATABASE
 */
Oid
createdb(ParseState *pstate, const CreatedbStmt *stmt)
{
 Oid   src_dboid;
 Oid   src_owner;
 int   src_encoding = -1;
 char    *src_collate = NULL;
 char    *src_ctype = NULL;
 char    *src_locale = NULL;
 char    *src_icurules = NULL;
 char  src_locprovider = '\0';
 char    *src_collversion = NULL;
 bool  src_istemplate;
 bool  src_hasloginevt = false;
 bool  src_allowconn;
 TransactionId src_frozenxid = InvalidTransactionId;
 MultiXactId src_minmxid = InvalidMultiXactId;
 Oid   src_deftablespace;
 volatile Oid dst_deftablespace;
 Relation pg_database_rel;
 HeapTuple tuple;
 Datum  new_record[Natts_pg_database] = {0};
 bool  new_record_nulls[Natts_pg_database] = {0};
 Oid   dboid = InvalidOid;
 Oid   datdba;
 ListCell   *option;
 DefElem    *tablespacenameEl = NULL;
 DefElem    *ownerEl = NULL;
 DefElem    *templateEl = NULL;
 DefElem    *encodingEl = NULL;
 DefElem    *localeEl = NULL;
 DefElem    *builtinlocaleEl = NULL;
 DefElem    *collateEl = NULL;
 DefElem    *ctypeEl = NULL;
 DefElem    *iculocaleEl = NULL;
 DefElem    *icurulesEl = NULL;
 DefElem    *locproviderEl = NULL;
 DefElem    *istemplateEl = NULL;
 DefElem    *allowconnectionsEl = NULL;
 DefElem    *connlimitEl = NULL;
 DefElem    *collversionEl = NULL;
 DefElem    *strategyEl = NULL;
 char    *dbname = stmt->dbname;
 char    *dbowner = NULL;
 const char *dbtemplate = NULL;
 char    *dbcollate = NULL;
 char    *dbctype = NULL;
 const char *dblocale = NULL;
 char    *dbicurules = NULL;
 char  dblocprovider = '\0';
 char    *canonname;
 int   encoding = -1;
 bool  dbistemplate = false;
 bool  dballowconnections = true;
 int   dbconnlimit = DATCONNLIMIT_UNLIMITED;
 char    *dbcollversion = NULL;
 int   notherbackends;
 int   npreparedxacts;
 CreateDBStrategy dbstrategy = CREATEDB_WAL_LOG;
 createdb_failure_params fparms;

 /* Extract options from the statement node tree */
 foreach(option, stmt->options)
 {
  DefElem    *defel = (DefElem *) lfirst(option);

  if (strcmp(defel->defname, "tablespace") == 0)
  {
   if (tablespacenameEl)
    errorConflictingDefElem(defel, pstate);
   tablespacenameEl = defel;
  }
  else if (strcmp(defel->defname, "owner") == 0)
  {
   if (ownerEl)
    errorConflictingDefElem(defel, pstate);
   ownerEl = defel;
  }
  else if (strcmp(defel->defname, "template") == 0)
  {
   if (templateEl)
    errorConflictingDefElem(defel, pstate);
   templateEl = defel;
  }
  else if (strcmp(defel->defname, "encoding") == 0)
  {
   if (encodingEl)
    errorConflictingDefElem(defel, pstate);
   encodingEl = defel;
  }
  else if (strcmp(defel->defname, "locale") == 0)
  {
   if (localeEl)
    errorConflictingDefElem(defel, pstate);
   localeEl = defel;
  }
  else if (strcmp(defel->defname, "builtin_locale") == 0)
  {
   if (builtinlocaleEl)
    errorConflictingDefElem(defel, pstate);
   builtinlocaleEl = defel;
  }
  else if (strcmp(defel->defname, "lc_collate") == 0)
  {
   if (collateEl)
    errorConflictingDefElem(defel, pstate);
   collateEl = defel;
  }
  else if (strcmp(defel->defname, "lc_ctype") == 0)
  {
   if (ctypeEl)
    errorConflictingDefElem(defel, pstate);
   ctypeEl = defel;
  }
  else if (strcmp(defel->defname, "icu_locale") == 0)
  {
   if (iculocaleEl)
    errorConflictingDefElem(defel, pstate);
   iculocaleEl = defel;
  }
  else if (strcmp(defel->defname, "icu_rules") == 0)
  {
   if (icurulesEl)
    errorConflictingDefElem(defel, pstate);
   icurulesEl = defel;
  }
  else if (strcmp(defel->defname, "locale_provider") == 0)
  {
   if (locproviderEl)
    errorConflictingDefElem(defel, pstate);
   locproviderEl = defel;
  }
  else if (strcmp(defel->defname, "is_template") == 0)
  {
   if (istemplateEl)
    errorConflictingDefElem(defel, pstate);
   istemplateEl = defel;
  }
  else if (strcmp(defel->defname, "allow_connections") == 0)
  {
   if (allowconnectionsEl)
    errorConflictingDefElem(defel, pstate);
   allowconnectionsEl = defel;
  }
  else if (strcmp(defel->defname, "connection_limit") == 0)
  {
   if (connlimitEl)
    errorConflictingDefElem(defel, pstate);
   connlimitEl = defel;
  }
  else if (strcmp(defel->defname, "collation_version") == 0)
  {
   if (collversionEl)
    errorConflictingDefElem(defel, pstate);
   collversionEl = defel;
  }
  else if (strcmp(defel->defname, "location") == 0)
  {
   ereport(WARNING,
     (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
      errmsg("LOCATION is not supported anymore"),
      errhint("Consider using tablespaces instead."),
      parser_errposition(pstate, defel->location)));
  }
  else if (strcmp(defel->defname, "oid") == 0)
  {
   dboid = defGetObjectId(defel);

   /*
    * We don't normally permit new databases to be created with
    * system-assigned OIDs. pg_upgrade tries to preserve database
    * OIDs, so we can't allow any database to be created with an OID
    * that might be in use in a freshly-initialized cluster created
    * by some future version. We assume all such OIDs will be from
    * the system-managed OID range.
    *
    * As an exception, however, we permit any OID to be assigned when
    * allow_system_table_mods=on (so that initdb can assign system
    * OIDs to template0 and postgres) or when performing a binary
    * upgrade (so that pg_upgrade can preserve whatever OIDs it finds
    * in the source cluster).
 */
   if (dboid < FirstNormalObjectId &&
    !allowSystemTableMods && !IsBinaryUpgrade)
    ereport(ERROR,
      (errcode(ERRCODE_INVALID_PARAMETER_VALUE)),
      errmsg("OIDs less than %u are reserved for system objects", FirstNormalObjectId));
  }
  else if (strcmp(defel->defname, "strategy") == 0)
  {
   if (strategyEl)
    errorConflictingDefElem(defel, pstate);
   strategyEl = defel;
  }
  else
   ereport(ERROR,
     (errcode(ERRCODE_SYNTAX_ERROR),
      errmsg("option \"%s\" not recognized", defel->defname),
      parser_errposition(pstate, defel->location)));
 }

 if (ownerEl && ownerEl->arg)
  dbowner = defGetString(ownerEl);
 if (templateEl && templateEl->arg)
  dbtemplate = defGetString(templateEl);
 if (encodingEl && encodingEl->arg)
 {
  const char *encoding_name;

  if (IsA(encodingEl->arg, Integer))
  {
   encoding = defGetInt32(encodingEl);
   encoding_name = pg_encoding_to_char(encoding);
   if (strcmp(encoding_name, "") == 0 ||
    pg_valid_server_encoding(encoding_name) < 0)
    ereport(ERROR,
      (errcode(ERRCODE_UNDEFINED_OBJECT),
       errmsg("%d is not a valid encoding code",
        encoding),
       parser_errposition(pstate, encodingEl->location)));
  }
  else
  {
   encoding_name = defGetString(encodingEl);
   encoding = pg_valid_server_encoding(encoding_name);
   if (encoding < 0)
    ereport(ERROR,
      (errcode(ERRCODE_UNDEFINED_OBJECT),
       errmsg("%s is not a valid encoding name",
        encoding_name),
       parser_errposition(pstate, encodingEl->location)));
  }
 }
 if (localeEl && localeEl->arg)
 {
  dbcollate = defGetString(localeEl);
  dbctype = defGetString(localeEl);
  dblocale = defGetString(localeEl);
 }
 if (builtinlocaleEl && builtinlocaleEl->arg)
  dblocale = defGetString(builtinlocaleEl);
 if (collateEl && collateEl->arg)
  dbcollate = defGetString(collateEl);
 if (ctypeEl && ctypeEl->arg)
  dbctype = defGetString(ctypeEl);
 if (iculocaleEl && iculocaleEl->arg)
  dblocale = defGetString(iculocaleEl);
 if (icurulesEl && icurulesEl->arg)
  dbicurules = defGetString(icurulesEl);
 if (locproviderEl && locproviderEl->arg)
 {
  char    *locproviderstr = defGetString(locproviderEl);

  if (pg_strcasecmp(locproviderstr, "builtin") == 0)
   dblocprovider = COLLPROVIDER_BUILTIN;
  else if (pg_strcasecmp(locproviderstr, "icu") == 0)
   dblocprovider = COLLPROVIDER_ICU;
  else if (pg_strcasecmp(locproviderstr, "libc") == 0)
   dblocprovider = COLLPROVIDER_LIBC;
  else
   ereport(ERROR,
     (errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
      errmsg("unrecognized locale provider: %s",
       locproviderstr)));
 }
 if (istemplateEl && istemplateEl->arg)
  dbistemplate = defGetBoolean(istemplateEl);
 if (allowconnectionsEl && allowconnectionsEl->arg)
  dballowconnections = defGetBoolean(allowconnectionsEl);
 if (connlimitEl && connlimitEl->arg)
 {
  dbconnlimit = defGetInt32(connlimitEl);
  if (dbconnlimit < DATCONNLIMIT_UNLIMITED)
   ereport(ERROR,
     (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
      errmsg("invalid connection limit: %d", dbconnlimit)));
 }
 if (collversionEl)
  dbcollversion = defGetString(collversionEl);

 /* obtain OID of proposed owner */
 if (dbowner)
  datdba = get_role_oid(dbowner, false);
 else
  datdba = GetUserId();

 /*
  * To create a database, must have createdb privilege and must be able to
  * become the target role (this does not imply that the target role itself
  * must have createdb privilege).  The latter provision guards against
  * "giveaway" attacks.  Note that a superuser will always have both of
  * these privileges a fortiori.
 */
 if (!have_createdb_privilege())
  ereport(ERROR,
    (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
     errmsg("permission denied to create database")));

 check_can_set_role(GetUserId(), datdba);

 /*
  * Lookup database (template) to be cloned, and obtain share lock on it.
  * ShareLock allows two CREATE DATABASEs to work from the same template
  * concurrently, while ensuring no one is busy dropping it in parallel
  * (which would be Very Bad since we'd likely get an incomplete copy
  * without knowing it).  This also prevents any new connections from being
  * made to the source until we finish copying it, so we can be sure it
  * won't change underneath us.
 */
 if (!dbtemplate)
  dbtemplate = "template1"; /* Default template database name */

 if (!get_db_info(dbtemplate, ShareLock,
      &src_dboid, &src_owner, &src_encoding,
      &src_istemplate, &src_allowconn, &src_hasloginevt,
      &src_frozenxid, &src_minmxid, &src_deftablespace,
      &src_collate, &src_ctype, &src_locale, &src_icurules, &src_locprovider,
      &src_collversion))
  ereport(ERROR,
    (errcode(ERRCODE_UNDEFINED_DATABASE),
     errmsg("template database \"%s\" does not exist",
      dbtemplate)));

 /*
  * If the source database was in the process of being dropped, we can't
  * use it as a template.
 */
 if (database_is_invalid_oid(src_dboid))
  ereport(ERROR,
    errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
    errmsg("cannot use invalid database \"%s\" as template", dbtemplate),
    errhint("Use DROP DATABASE to drop invalid databases."));

 /*
  * Permission check: to copy a DB that's not marked datistemplate, you
  * must be superuser or the owner thereof.
 */
 if (!src_istemplate)
 {
  if (!object_ownercheck(DatabaseRelationId, src_dboid, GetUserId()))
   ereport(ERROR,
     (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
      errmsg("permission denied to copy database \"%s\"",
       dbtemplate)));
 }

 /* Validate the database creation strategy. */
 if (strategyEl && strategyEl->arg)
 {
  char    *strategy;

  strategy = defGetString(strategyEl);
  if (pg_strcasecmp(strategy, "wal_log") == 0)
   dbstrategy = CREATEDB_WAL_LOG;
  else if (pg_strcasecmp(strategy, "file_copy") == 0)
   dbstrategy = CREATEDB_FILE_COPY;
  else
   ereport(ERROR,
     (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
      errmsg("invalid create database strategy \"%s\"", strategy),
      errhint("Valid strategies are \"wal_log\" and \"file_copy\".")));
 }

 /* If encoding or locales are defaulted, use source's setting */
 if (encoding < 0)
  encoding = src_encoding;
 if (dbcollate == NULL)
  dbcollate = src_collate;
 if (dbctype == NULL)
  dbctype = src_ctype;
 if (dblocprovider == '\0')
  dblocprovider = src_locprovider;
 if (dblocale == NULL && dblocprovider == src_locprovider)
  dblocale = src_locale;
 if (dbicurules == NULL)
  dbicurules = src_icurules;

 /* Some encodings are client only */
 if (!PG_VALID_BE_ENCODING(encoding))
  ereport(ERROR,
    (errcode(ERRCODE_WRONG_OBJECT_TYPE),
     errmsg("invalid server encoding %d", encoding)));

 /* Check that the chosen locales are valid, and get canonical spellings */
 if (!check_locale(LC_COLLATE, dbcollate, &canonname))
 {
  if (dblocprovider == COLLPROVIDER_BUILTIN)
   ereport(ERROR,
     (errcode(ERRCODE_WRONG_OBJECT_TYPE),
      errmsg("invalid LC_COLLATE locale name: \"%s\"", dbcollate),
      errhint("If the locale name is specific to the builtin provider, use BUILTIN_LOCALE.")));
  else if (dblocprovider == COLLPROVIDER_ICU)
   ereport(ERROR,
     (errcode(ERRCODE_WRONG_OBJECT_TYPE),
      errmsg("invalid LC_COLLATE locale name: \"%s\"", dbcollate),
      errhint("If the locale name is specific to the ICU provider, use ICU_LOCALE.")));
  else
   ereport(ERROR,
     (errcode(ERRCODE_WRONG_OBJECT_TYPE),
      errmsg("invalid LC_COLLATE locale name: \"%s\"", dbcollate)));
 }
 dbcollate = canonname;
 if (!check_locale(LC_CTYPE, dbctype, &canonname))
 {
  if (dblocprovider == COLLPROVIDER_BUILTIN)
   ereport(ERROR,
     (errcode(ERRCODE_WRONG_OBJECT_TYPE),
      errmsg("invalid LC_CTYPE locale name: \"%s\"", dbctype),
      errhint("If the locale name is specific to the builtin provider, use BUILTIN_LOCALE.")));
  else if (dblocprovider == COLLPROVIDER_ICU)
   ereport(ERROR,
     (errcode(ERRCODE_WRONG_OBJECT_TYPE),
      errmsg("invalid LC_CTYPE locale name: \"%s\"", dbctype),
      errhint("If the locale name is specific to the ICU provider, use ICU_LOCALE.")));
  else
   ereport(ERROR,
     (errcode(ERRCODE_WRONG_OBJECT_TYPE),
      errmsg("invalid LC_CTYPE locale name: \"%s\"", dbctype)));
 }

 dbctype = canonname;

 check_encoding_locale_matches(encoding, dbcollate, dbctype);

 /* validate provider-specific parameters */
 if (dblocprovider != COLLPROVIDER_BUILTIN)
 {
  if (builtinlocaleEl)
   ereport(ERROR,
     (errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
      errmsg("BUILTIN_LOCALE cannot be specified unless locale provider is builtin")));
 }

 if (dblocprovider != COLLPROVIDER_ICU)
 {
  if (iculocaleEl)
   ereport(ERROR,
     (errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
      errmsg("ICU locale cannot be specified unless locale provider is ICU")));

  if (dbicurules)
   ereport(ERROR,
     (errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
      errmsg("ICU rules cannot be specified unless locale provider is ICU")));
 }

 /* validate and canonicalize locale for the provider */
 if (dblocprovider == COLLPROVIDER_BUILTIN)
 {
  /*
   * This would happen if template0 uses the libc provider but the new
   * database uses builtin.
 */
  if (!dblocale)
   ereport(ERROR,
     (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
      errmsg("LOCALE or BUILTIN_LOCALE must be specified")));

  dblocale = builtin_validate_locale(encoding, dblocale);
 }
 else if (dblocprovider == COLLPROVIDER_ICU)
 {
  if (!(is_encoding_supported_by_icu(encoding)))
   ereport(ERROR,
     (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
      errmsg("encoding \"%s\" is not supported with ICU provider",
       pg_encoding_to_char(encoding))));

  /*
   * This would happen if template0 uses the libc provider but the new
   * database uses icu.
 */
  if (!dblocale)
   ereport(ERROR,
     (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
      errmsg("LOCALE or ICU_LOCALE must be specified")));

  /*
   * During binary upgrade, or when the locale came from the template
   * database, preserve locale string. Otherwise, canonicalize to a
   * language tag.
 */
  if (!IsBinaryUpgrade && dblocale != src_locale)
  {
   char    *langtag = icu_language_tag(dblocale,
               icu_validation_level);

   if (langtag && strcmp(dblocale, langtag) != 0)
   {
    ereport(NOTICE,
      (errmsg("using standard form \"%s\" for ICU locale \"%s\"",
        langtag, dblocale)));

    dblocale = langtag;
   }
  }

  icu_validate_locale(dblocale);
 }

 /* for libc, locale comes from datcollate and datctype */
 if (dblocprovider == COLLPROVIDER_LIBC)
  dblocale = NULL;

 /*
  * Check that the new encoding and locale settings match the source
  * database.  We insist on this because we simply copy the source data ---
  * any non-ASCII data would be wrongly encoded, and any indexes sorted
  * according to the source locale would be wrong.
  *
  * However, we assume that template0 doesn't contain any non-ASCII data
  * nor any indexes that depend on collation or ctype, so template0 can be
  * used as template for creating a database with any encoding or locale.
 */
 if (strcmp(dbtemplate, "template0") != 0)
 {
  if (encoding != src_encoding)
   ereport(ERROR,
     (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
      errmsg("new encoding (%s) is incompatible with the encoding of the template database (%s)",
       pg_encoding_to_char(encoding),
       pg_encoding_to_char(src_encoding)),
      errhint("Use the same encoding as in the template database, or use template0 as template.")));

  if (strcmp(dbcollate, src_collate) != 0)
   ereport(ERROR,
     (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
      errmsg("new collation (%s) is incompatible with the collation of the template database (%s)",
       dbcollate, src_collate),
      errhint("Use the same collation as in the template database, or use template0 as template.")));

  if (strcmp(dbctype, src_ctype) != 0)
   ereport(ERROR,
     (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
      errmsg("new LC_CTYPE (%s) is incompatible with the LC_CTYPE of the template database (%s)",
       dbctype, src_ctype),
      errhint("Use the same LC_CTYPE as in the template database, or use template0 as template.")));

  if (dblocprovider != src_locprovider)
   ereport(ERROR,
     (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
      errmsg("new locale provider (%s) does not match locale provider of the template database (%s)",
       collprovider_name(dblocprovider), collprovider_name(src_locprovider)),
      errhint("Use the same locale provider as in the template database, or use template0 as template.")));

  if (dblocprovider == COLLPROVIDER_ICU)
  {
   char    *val1;
   char    *val2;

   Assert(dblocale);
   Assert(src_locale);
   if (strcmp(dblocale, src_locale) != 0)
    ereport(ERROR,
      (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
       errmsg("new ICU locale (%s) is incompatible with the ICU locale of the template database (%s)",
        dblocale, src_locale),
       errhint("Use the same ICU locale as in the template database, or use template0 as template.")));

   val1 = dbicurules;
   if (!val1)
    val1 = "";
   val2 = src_icurules;
   if (!val2)
    val2 = "";
   if (strcmp(val1, val2) != 0)
    ereport(ERROR,
      (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
       errmsg("new ICU collation rules (%s) are incompatible with the ICU collation rules of the template database (%s)",
        val1, val2),
       errhint("Use the same ICU collation rules as in the template database, or use template0 as template.")));
  }
 }

 /*
  * If we got a collation version for the template database, check that it
  * matches the actual OS collation version.  Otherwise error; the user
  * needs to fix the template database first.  Don't complain if a
  * collation version was specified explicitly as a statement option; that
  * is used by pg_upgrade to reproduce the old state exactly.
  *
  * (If the template database has no collation version, then either the
  * platform/provider does not support collation versioning, or it's
  * template0, for which we stipulate that it does not contain
  * collation-using objects.)
 */
 if (src_collversion && !collversionEl)
 {
  char    *actual_versionstr;
  const char *locale;

  if (dblocprovider == COLLPROVIDER_LIBC)
   locale = dbcollate;
  else
   locale = dblocale;

  actual_versionstr = get_collation_actual_version(dblocprovider, locale);
  if (!actual_versionstr)
   ereport(ERROR,
     (errmsg("template database \"%s\" has a collation version, but no actual collation version could be determined",
       dbtemplate)));

  if (strcmp(actual_versionstr, src_collversion) != 0)
   ereport(ERROR,
     (errmsg("template database \"%s\" has a collation version mismatch",
       dbtemplate),
      errdetail("The template database was created using collation version %s, "
          "but the operating system provides version %s.",
          src_collversion, actual_versionstr),
      errhint("Rebuild all objects in the template database that use the default collation and run "
        "ALTER DATABASE %s REFRESH COLLATION VERSION, "
        "or build PostgreSQL with the right library version.",
        quote_identifier(dbtemplate))));
 }

 if (dbcollversion == NULL)
  dbcollversion = src_collversion;

 /*
  * Normally, we copy the collation version from the template database.
  * This last resort only applies if the template database does not have a
  * collation version, which is normally only the case for template0.
 */
 if (dbcollversion == NULL)
 {
  const char *locale;

  if (dblocprovider == COLLPROVIDER_LIBC)
   locale = dbcollate;
  else
   locale = dblocale;

  dbcollversion = get_collation_actual_version(dblocprovider, locale);
 }

 /* Resolve default tablespace for new database */
 if (tablespacenameEl && tablespacenameEl->arg)
 {
  char    *tablespacename;
  AclResult aclresult;

  tablespacename = defGetString(tablespacenameEl);
  dst_deftablespace = get_tablespace_oid(tablespacename, false);
  /* check permissions */
  aclresult = object_aclcheck(TableSpaceRelationId, dst_deftablespace, GetUserId(),
         ACL_CREATE);
  if (aclresult != ACLCHECK_OK)
   aclcheck_error(aclresult, OBJECT_TABLESPACE,
         tablespacename);

  /* pg_global must never be the default tablespace */
  if (dst_deftablespace == GLOBALTABLESPACE_OID)
   ereport(ERROR,
     (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
      errmsg("pg_global cannot be used as default tablespace")));

  /*
   * If we are trying to change the default tablespace of the template,
   * we require that the template not have any files in the new default
   * tablespace.  This is necessary because otherwise the copied
   * database would contain pg_class rows that refer to its default
   * tablespace both explicitly (by OID) and implicitly (as zero), which
   * would cause problems.  For example another CREATE DATABASE using
   * the copied database as template, and trying to change its default
   * tablespace again, would yield outright incorrect results (it would
   * improperly move tables to the new default tablespace that should
   * stay in the same tablespace).
 */
  if (dst_deftablespace != src_deftablespace)
  {
   char    *srcpath;
   struct stat st;

   srcpath = GetDatabasePath(src_dboid, dst_deftablespace);

   if (stat(srcpath, &st) == 0 &&
    S_ISDIR(st.st_mode) &&
    !directory_is_empty(srcpath))
    ereport(ERROR,
      (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
       errmsg("cannot assign new default tablespace \"%s\"",
        tablespacename),
       errdetail("There is a conflict because database \"%s\" already has some tables in this tablespace.",
           dbtemplate)));
   pfree(srcpath);
  }
 }
 else
 {
  /* Use template database's default tablespace */
  dst_deftablespace = src_deftablespace;
  /* Note there is no additional permission check in this path */
 }

 /*
  * If built with appropriate switch, whine when regression-testing
  * conventions for database names are violated.  But don't complain during
  * initdb.
 */
#ifdef ENFORCE_REGRESSION_TEST_NAME_RESTRICTIONS
 if (IsUnderPostmaster && strstr(dbname, "regression") == NULL)
  elog(WARNING, "databases created by regression test cases should have names including \"regression\"");
#endif

 /*
  * Check for db name conflict.  This is just to give a more friendly error
  * message than "unique index violation".  There's a race condition but
  * we're willing to accept the less friendly message in that case.
 */
 if (OidIsValid(get_database_oid(dbname, true)))
  ereport(ERROR,
    (errcode(ERRCODE_DUPLICATE_DATABASE),
     errmsg("database \"%s\" already exists", dbname)));

 /*
  * The source DB can't have any active backends, except this one
  * (exception is to allow CREATE DB while connected to template1).
  * Otherwise we might copy inconsistent data.
  *
  * This should be last among the basic error checks, because it involves
  * potential waiting; we may as well throw an error first if we're gonna
  * throw one.
 */
 if (CountOtherDBBackends(src_dboid, ¬herbackends, &npreparedxacts))
  ereport(ERROR,
    (errcode(ERRCODE_OBJECT_IN_USE),
     errmsg("source database \"%s\" is being accessed by other users",
      dbtemplate),
     errdetail_busy_db(notherbackends, npreparedxacts)));

 /*
  * Select an OID for the new database, checking that it doesn't have a
  * filename conflict with anything already existing in the tablespace
  * directories.
 */
 pg_database_rel = table_open(DatabaseRelationId, RowExclusiveLock);

 /*
  * If database OID is configured, check if the OID is already in use or
  * data directory already exists.
 */
 if (OidIsValid(dboid))
 {
  char    *existing_dbname = get_database_name(dboid);

  if (existing_dbname != NULL)
   ereport(ERROR,
     (errcode(ERRCODE_INVALID_PARAMETER_VALUE)),
     errmsg("database OID %u is already in use by database \"%s\"",
         dboid, existing_dbname));

  if (check_db_file_conflict(dboid))
   ereport(ERROR,
     (errcode(ERRCODE_INVALID_PARAMETER_VALUE)),
     errmsg("data directory with the specified OID %u already exists", dboid));
 }
 else
 {
  /* Select an OID for the new database if is not explicitly configured. */
  do
  {
   dboid = GetNewOidWithIndex(pg_database_rel, DatabaseOidIndexId,
            Anum_pg_database_oid);
  } while (check_db_file_conflict(dboid));
 }

 /*
  * Insert a new tuple into pg_database.  This establishes our ownership of
  * the new database name (anyone else trying to insert the same name will
  * block on the unique index, and fail after we commit).
 */

 Assert((dblocprovider != COLLPROVIDER_LIBC && dblocale) ||
     (dblocprovider == COLLPROVIDER_LIBC && !dblocale));

 /* Form tuple */
 new_record[Anum_pg_database_oid - 1] = ObjectIdGetDatum(dboid);
 new_record[Anum_pg_database_datname - 1] =
  DirectFunctionCall1(namein, CStringGetDatum(dbname));
 new_record[Anum_pg_database_datdba - 1] = ObjectIdGetDatum(datdba);
 new_record[Anum_pg_database_encoding - 1] = Int32GetDatum(encoding);
 new_record[Anum_pg_database_datlocprovider - 1] = CharGetDatum(dblocprovider);
 new_record[Anum_pg_database_datistemplate - 1] = BoolGetDatum(dbistemplate);
 new_record[Anum_pg_database_datallowconn - 1] = BoolGetDatum(dballowconnections);
 new_record[Anum_pg_database_dathasloginevt - 1] = BoolGetDatum(src_hasloginevt);
 new_record[Anum_pg_database_datconnlimit - 1] = Int32GetDatum(dbconnlimit);
 new_record[Anum_pg_database_datfrozenxid - 1] = TransactionIdGetDatum(src_frozenxid);
 new_record[Anum_pg_database_datminmxid - 1] = TransactionIdGetDatum(src_minmxid);
 new_record[Anum_pg_database_dattablespace - 1] = ObjectIdGetDatum(dst_deftablespace);
 new_record[Anum_pg_database_datcollate - 1] = CStringGetTextDatum(dbcollate);
 new_record[Anum_pg_database_datctype - 1] = CStringGetTextDatum(dbctype);
 if (dblocale)
  new_record[Anum_pg_database_datlocale - 1] = CStringGetTextDatum(dblocale);
 else
  new_record_nulls[Anum_pg_database_datlocale - 1] = true;
 if (dbicurules)
  new_record[Anum_pg_database_daticurules - 1] = CStringGetTextDatum(dbicurules);
 else
  new_record_nulls[Anum_pg_database_daticurules - 1] = true;
 if (dbcollversion)
  new_record[Anum_pg_database_datcollversion - 1] = CStringGetTextDatum(dbcollversion);
 else
  new_record_nulls[Anum_pg_database_datcollversion - 1] = true;

 /*
  * We deliberately set datacl to default (NULL), rather than copying it
  * from the template database.  Copying it would be a bad idea when the
  * owner is not the same as the template's owner.
 */
 new_record_nulls[Anum_pg_database_datacl - 1] = true;

 tuple = heap_form_tuple(RelationGetDescr(pg_database_rel),
       new_record, new_record_nulls);

 CatalogTupleInsert(pg_database_rel, tuple);

 /*
  * Now generate additional catalog entries associated with the new DB
 */

 /* Register owner dependency */
 recordDependencyOnOwner(DatabaseRelationId, dboid, datdba);

 /* Create pg_shdepend entries for objects within database */
 copyTemplateDependencies(src_dboid, dboid);

 /* Post creation hook for new database */
 InvokeObjectPostCreateHook(DatabaseRelationId, dboid, 0);

 /*
  * If we're going to be reading data for the to-be-created database into
  * shared_buffers, take a lock on it. Nobody should know that this
  * database exists yet, but it's good to maintain the invariant that an
  * AccessExclusiveLock on the database is sufficient to drop all of its
  * buffers without worrying about more being read later.
  *
  * Note that we need to do this before entering the
  * PG_ENSURE_ERROR_CLEANUP block below, because createdb_failure_callback
  * expects this lock to be held already.
 */
 if (dbstrategy == CREATEDB_WAL_LOG)
  LockSharedObject(DatabaseRelationId, dboid, 0, AccessShareLock);

 /*
  * Once we start copying subdirectories, we need to be able to clean 'em
  * up if we fail.  Use an ENSURE block to make sure this happens.  (This
  * is not a 100% solution, because of the possibility of failure during
  * transaction commit after we leave this routine, but it should handle
  * most scenarios.)
 */
 fparms.src_dboid = src_dboid;
 fparms.dest_dboid = dboid;
 fparms.strategy = dbstrategy;

 PG_ENSURE_ERROR_CLEANUP(createdb_failure_callback,
       PointerGetDatum(&fparms));
 {
  /*
   * If the user has asked to create a database with WAL_LOG strategy
   * then call CreateDatabaseUsingWalLog, which will copy the database
   * at the block level and it will WAL log each copied block.
   * Otherwise, call CreateDatabaseUsingFileCopy that will copy the
   * database file by file.
 */
  if (dbstrategy == CREATEDB_WAL_LOG)
   CreateDatabaseUsingWalLog(src_dboid, dboid, src_deftablespace,
           dst_deftablespace);
  else
   CreateDatabaseUsingFileCopy(src_dboid, dboid, src_deftablespace,
          dst_deftablespace);

  /*
   * Close pg_database, but keep lock till commit.
 */
  table_close(pg_database_rel, NoLock);

  /*
   * Force synchronous commit, thus minimizing the window between
   * creation of the database files and committal of the transaction. If
   * we crash before committing, we'll have a DB that's taking up disk
   * space but is not in pg_database, which is not good.
 */
  ForceSyncCommit();
 }
 PG_END_ENSURE_ERROR_CLEANUP(createdb_failure_callback,
        PointerGetDatum(&fparms));

 return dboid;
}

/*
 * Check whether chosen encoding matches chosen locale settings.  This
 * restriction is necessary because libc's locale-specific code usually
 * fails when presented with data in an encoding it's not expecting. We
 * allow mismatch in four cases:
 *
 * 1. locale encoding = SQL_ASCII, which means that the locale is C/POSIX
 * which works with any encoding.
 *
 * 2. locale encoding = -1, which means that we couldn't determine the
 * locale's encoding and have to trust the user to get it right.
 *
 * 3. selected encoding is UTF8 and platform is win32. This is because
 * UTF8 is a pseudo codepage that is supported in all locales since it's
 * converted to UTF16 before being used.
 *
 * 4. selected encoding is SQL_ASCII, but only if you're a superuser. This
 * is risky but we have historically allowed it --- notably, the
 * regression tests require it.
 *
 * Note: if you change this policy, fix initdb to match.
 */
void
check_encoding_locale_matches(int encoding, const char *collate, const char *ctype)
{
 int   ctype_encoding = pg_get_encoding_from_locale(ctype, true);
 int   collate_encoding = pg_get_encoding_from_locale(collate, true);

 if (!(ctype_encoding == encoding ||
    ctype_encoding == PG_SQL_ASCII ||
    ctype_encoding == -1 ||
#ifdef WIN32
    encoding == PG_UTF8 ||
#endif
    (encoding == PG_SQL_ASCII && superuser())))
  ereport(ERROR,
    (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
     errmsg("encoding \"%s\" does not match locale \"%s\"",
      pg_encoding_to_char(encoding),
      ctype),
     errdetail("The chosen LC_CTYPE setting requires encoding \"%s\".",
         pg_encoding_to_char(ctype_encoding))));

 if (!(collate_encoding == encoding ||
    collate_encoding == PG_SQL_ASCII ||
    collate_encoding == -1 ||
#ifdef WIN32
    encoding == PG_UTF8 ||
#endif
    (encoding == PG_SQL_ASCII && superuser())))
  ereport(ERROR,
    (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
     errmsg("encoding \"%s\" does not match locale \"%s\"",
      pg_encoding_to_char(encoding),
      collate),
     errdetail("The chosen LC_COLLATE setting requires encoding \"%s\".",
         pg_encoding_to_char(collate_encoding))));
}

/* Error cleanup callback for createdb */
static void
createdb_failure_callback(int code, Datum arg)
{
 createdb_failure_params *fparms = (createdb_failure_params *) DatumGetPointer(arg);

 /*
  * If we were copying database at block levels then drop pages for the
  * destination database that are in the shared buffer cache.  And tell
  * checkpointer to forget any pending fsync and unlink requests for files
  * in the database.  The reasoning behind doing this is same as explained
  * in dropdb function.  But unlike dropdb we don't need to call
  * pgstat_drop_database because this database is still not created so
  * there should not be any stat for this.
 */
 if (fparms->strategy == CREATEDB_WAL_LOG)
 {
  DropDatabaseBuffers(fparms->dest_dboid);
  ForgetDatabaseSyncRequests(fparms->dest_dboid);

  /* Release lock on the target database. */
  UnlockSharedObject(DatabaseRelationId, fparms->dest_dboid, 0,
         AccessShareLock);
 }

 /*
  * Release lock on source database before doing recursive remove. This is
  * not essential but it seems desirable to release the lock as soon as
  * possible.
 */
 UnlockSharedObject(DatabaseRelationId, fparms->src_dboid, 0, ShareLock);

 /* Throw away any successfully copied subdirectories */
 remove_dbtablespaces(fparms->dest_dboid);
}


/*
 * DROP DATABASE
 */
void
dropdb(const char *dbname, bool missing_ok, bool force)
{
 Oid   db_id;
 bool  db_istemplate;
 Relation pgdbrel;
 HeapTuple tup;
 ScanKeyData scankey;
 void    *inplace_state;
 Form_pg_database datform;
 int   notherbackends;
 int   npreparedxacts;
 int   nslots,
    nslots_active;
 int   nsubscriptions;

 /*
  * Look up the target database's OID, and get exclusive lock on it. We
  * need this to ensure that no new backend starts up in the target
  * database while we are deleting it (see postinit.c), and that no one is
  * using it as a CREATE DATABASE template or trying to delete it for
  * themselves.
 */
 pgdbrel = table_open(DatabaseRelationId, RowExclusiveLock);

 if (!get_db_info(dbname, AccessExclusiveLock, &db_id, NULL, NULL,
      &db_istemplate, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL))
 {
  if (!missing_ok)
  {
   ereport(ERROR,
     (errcode(ERRCODE_UNDEFINED_DATABASE),
      errmsg("database \"%s\" does not exist", dbname)));
  }
  else
  {
   /* Close pg_database, release the lock, since we changed nothing */
   table_close(pgdbrel, RowExclusiveLock);
   ereport(NOTICE,
     (errmsg("database \"%s\" does not exist, skipping",
       dbname)));
   return;
  }
 }

 /*
  * Permission checks
 */
 if (!object_ownercheck(DatabaseRelationId, db_id, GetUserId()))
  aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_DATABASE,
        dbname);

 /* DROP hook for the database being removed */
 InvokeObjectDropHook(DatabaseRelationId, db_id, 0);

 /*
  * Disallow dropping a DB that is marked istemplate.  This is just to
  * prevent people from accidentally dropping template0 or template1; they
  * can do so if they're really determined ...
 */
 if (db_istemplate)
  ereport(ERROR,
    (errcode(ERRCODE_WRONG_OBJECT_TYPE),
     errmsg("cannot drop a template database")));

 /* Obviously can't drop my own database */
 if (db_id == MyDatabaseId)
  ereport(ERROR,
    (errcode(ERRCODE_OBJECT_IN_USE),
     errmsg("cannot drop the currently open database")));

 /*
  * Check whether there are active logical slots that refer to the
  * to-be-dropped database. The database lock we are holding prevents the
  * creation of new slots using the database or existing slots becoming
  * active.
 */
 (void) ReplicationSlotsCountDBSlots(db_id, &nslots, &nslots_active);
 if (nslots_active)
 {
  ereport(ERROR,
    (errcode(ERRCODE_OBJECT_IN_USE),
     errmsg("database \"%s\" is used by an active logical replication slot",
      dbname),
     errdetail_plural("There is %d active slot.",
          "There are %d active slots.",
          nslots_active, nslots_active)));
 }

 /*
  * Check if there are subscriptions defined in the target database.
  *
  * We can't drop them automatically because they might be holding
  * resources in other databases/instances.
 */
 if ((nsubscriptions = CountDBSubscriptions(db_id)) > 0)
  ereport(ERROR,
    (errcode(ERRCODE_OBJECT_IN_USE),
     errmsg("database \"%s\" is being used by logical replication subscription",
      dbname),
     errdetail_plural("There is %d subscription.",
          "There are %d subscriptions.",
          nsubscriptions, nsubscriptions)));


 /*
  * Attempt to terminate all existing connections to the target database if
  * the user has requested to do so.
 */
 if (force)
  TerminateOtherDBBackends(db_id);

 /*
  * Check for other backends in the target database.  (Because we hold the
  * database lock, no new ones can start after this.)
  *
  * As in CREATE DATABASE, check this after other error conditions.
 */
 if (CountOtherDBBackends(db_id, ¬herbackends, &npreparedxacts))
  ereport(ERROR,
    (errcode(ERRCODE_OBJECT_IN_USE),
     errmsg("database \"%s\" is being accessed by other users",
      dbname),
     errdetail_busy_db(notherbackends, npreparedxacts)));

 /*
  * Delete any comments or security labels associated with the database.
 */
 DeleteSharedComments(db_id, DatabaseRelationId);
 DeleteSharedSecurityLabel(db_id, DatabaseRelationId);

 /*
  * Remove settings associated with this database
 */
 DropSetting(db_id, InvalidOid);

 /*
  * Remove shared dependency references for the database.
 */
 dropDatabaseDependencies(db_id);

 /*
  * Tell the cumulative stats system to forget it immediately, too.
 */
 pgstat_drop_database(db_id);

 /*
  * Except for the deletion of the catalog row, subsequent actions are not
  * transactional (consider DropDatabaseBuffers() discarding modified
  * buffers). But we might crash or get interrupted below. To prevent
  * accesses to a database with invalid contents, mark the database as
  * invalid using an in-place update.
  *
  * We need to flush the WAL before continuing, to guarantee the
  * modification is durable before performing irreversible filesystem
  * operations.
 */
 ScanKeyInit(&scankey,
    Anum_pg_database_datname,
    BTEqualStrategyNumber, F_NAMEEQ,
    CStringGetDatum(dbname));
 systable_inplace_update_begin(pgdbrel, DatabaseNameIndexId, true,
          NULL, 1, &scankey, &tup, &inplace_state);
 if (!HeapTupleIsValid(tup))
  elog(ERROR, "cache lookup failed for database %u", db_id);
 datform = (Form_pg_database) GETSTRUCT(tup);
 datform->datconnlimit = DATCONNLIMIT_INVALID_DB;
 systable_inplace_update_finish(inplace_state, tup);
 XLogFlush(XactLastRecEnd);

 /*
  * Also delete the tuple - transactionally. If this transaction commits,
  * the row will be gone, but if we fail, dropdb() can be invoked again.
 */
 CatalogTupleDelete(pgdbrel, &tup->t_self);
 heap_freetuple(tup);

 /*
  * Drop db-specific replication slots.
 */
 ReplicationSlotsDropDBSlots(db_id);

 /*
  * Drop pages for this database that are in the shared buffer cache. This
  * is important to ensure that no remaining backend tries to write out a
  * dirty buffer to the dead database later...
 */
 DropDatabaseBuffers(db_id);

 /*
  * Tell checkpointer to forget any pending fsync and unlink requests for
  * files in the database; else the fsyncs will fail at next checkpoint, or
  * worse, it will delete files that belong to a newly created database
  * with the same OID.
 */
 ForgetDatabaseSyncRequests(db_id);

 /*
  * Force a checkpoint to make sure the checkpointer has received the
  * message sent by ForgetDatabaseSyncRequests.
 */
 RequestCheckpoint(CHECKPOINT_IMMEDIATE | CHECKPOINT_FORCE | CHECKPOINT_WAIT);

 /* Close all smgr fds in all backends. */
 WaitForProcSignalBarrier(EmitProcSignalBarrier(PROCSIGNAL_BARRIER_SMGRRELEASE));

 /*
  * Remove all tablespace subdirs belonging to the database.
 */
 remove_dbtablespaces(db_id);

 /*
  * Close pg_database, but keep lock till commit.
 */
 table_close(pgdbrel, NoLock);

 /*
  * Force synchronous commit, thus minimizing the window between removal of
  * the database files and committal of the transaction. If we crash before
  * committing, we'll have a DB that's gone on disk but still there
  * according to pg_database, which is not good.
 */
 ForceSyncCommit();
}


/*
 * Rename database
 */
ObjectAddress
RenameDatabase(const char *oldname, const char *newname)
{
 Oid   db_id;
 HeapTuple newtup;
 ItemPointerData otid;
 Relation rel;
 int   notherbackends;
 int   npreparedxacts;
 ObjectAddress address;

 /*
  * Look up the target database's OID, and get exclusive lock on it. We
  * need this for the same reasons as DROP DATABASE.
 */
 rel = table_open(DatabaseRelationId, RowExclusiveLock);

 if (!get_db_info(oldname, AccessExclusiveLock, &db_id, NULL, NULL, NULL,
      NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL))
  ereport(ERROR,
    (errcode(ERRCODE_UNDEFINED_DATABASE),
     errmsg("database \"%s\" does not exist", oldname)));

 /* must be owner */
 if (!object_ownercheck(DatabaseRelationId, db_id, GetUserId()))
  aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_DATABASE,
        oldname);

 /* must have createdb rights */
 if (!have_createdb_privilege())
  ereport(ERROR,
    (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
     errmsg("permission denied to rename database")));

 /*
  * If built with appropriate switch, whine when regression-testing
  * conventions for database names are violated.
 */
#ifdef ENFORCE_REGRESSION_TEST_NAME_RESTRICTIONS
 if (strstr(newname, "regression") == NULL)
  elog(WARNING, "databases created by regression test cases should have names including \"regression\"");
#endif

 /*
  * Make sure the new name doesn't exist.  See notes for same error in
  * CREATE DATABASE.
 */
 if (OidIsValid(get_database_oid(newname, true)))
  ereport(ERROR,
    (errcode(ERRCODE_DUPLICATE_DATABASE),
     errmsg("database \"%s\" already exists", newname)));

 /*
  * XXX Client applications probably store the current database somewhere,
  * so renaming it could cause confusion.  On the other hand, there may not
  * be an actual problem besides a little confusion, so think about this
  * and decide.
 */
 if (db_id == MyDatabaseId)
  ereport(ERROR,
    (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
     errmsg("current database cannot be renamed")));

 /*
  * Make sure the database does not have active sessions.  This is the same
  * concern as above, but applied to other sessions.
  *
  * As in CREATE DATABASE, check this after other error conditions.
 */
 if (CountOtherDBBackends(db_id, ¬herbackends, &npreparedxacts))
  ereport(ERROR,
    (errcode(ERRCODE_OBJECT_IN_USE),
     errmsg("database \"%s\" is being accessed by other users",
      oldname),
     errdetail_busy_db(notherbackends, npreparedxacts)));

 /* rename */
 newtup = SearchSysCacheLockedCopy1(DATABASEOID, ObjectIdGetDatum(db_id));
 if (!HeapTupleIsValid(newtup))
  elog(ERROR, "cache lookup failed for database %u", db_id);
 otid = newtup->t_self;
 namestrcpy(&(((Form_pg_database) GETSTRUCT(newtup))->datname), newname);
 CatalogTupleUpdate(rel, &otid, newtup);
 UnlockTuple(rel, &otid, InplaceUpdateTupleLock);

 InvokeObjectPostAlterHook(DatabaseRelationId, db_id, 0);

 ObjectAddressSet(address, DatabaseRelationId, db_id);

 /*
  * Close pg_database, but keep lock till commit.
 */
 table_close(rel, NoLock);

 return address;
}


/*
 * ALTER DATABASE SET TABLESPACE
 */
static void
movedb(const char *dbname, const char *tblspcname)
{
 Oid   db_id;
 Relation pgdbrel;
 int   notherbackends;
 int   npreparedxacts;
 HeapTuple oldtuple,
    newtuple;
 Oid   src_tblspcoid,
    dst_tblspcoid;
 ScanKeyData scankey;
 SysScanDesc sysscan;
 AclResult aclresult;
 char    *src_dbpath;
 char    *dst_dbpath;
 DIR     *dstdir;
 struct dirent *xlde;
 movedb_failure_params fparms;

 /*
  * Look up the target database's OID, and get exclusive lock on it. We
  * need this to ensure that no new backend starts up in the database while
  * we are moving it, and that no one is using it as a CREATE DATABASE
  * template or trying to delete it.
 */
 pgdbrel = table_open(DatabaseRelationId, RowExclusiveLock);

 if (!get_db_info(dbname, AccessExclusiveLock, &db_id, NULL, NULL, NULL,
      NULL, NULL, NULL, NULL, &src_tblspcoid, NULL, NULL, NULL, NULL, NULL, NULL))
  ereport(ERROR,
    (errcode(ERRCODE_UNDEFINED_DATABASE),
     errmsg("database \"%s\" does not exist", dbname)));

 /*
  * We actually need a session lock, so that the lock will persist across
  * the commit/restart below.  (We could almost get away with letting the
  * lock be released at commit, except that someone could try to move
  * relations of the DB back into the old directory while we rmtree() it.)
 */
 LockSharedObjectForSession(DatabaseRelationId, db_id, 0,
          AccessExclusiveLock);

 /*
  * Permission checks
 */
 if (!object_ownercheck(DatabaseRelationId, db_id, GetUserId()))
  aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_DATABASE,
        dbname);

 /*
  * Obviously can't move the tables of my own database
 */
 if (db_id == MyDatabaseId)
  ereport(ERROR,
    (errcode(ERRCODE_OBJECT_IN_USE),
     errmsg("cannot change the tablespace of the currently open database")));

 /*
  * Get tablespace's oid
 */
 dst_tblspcoid = get_tablespace_oid(tblspcname, false);

 /*
  * Permission checks
 */
 aclresult = object_aclcheck(TableSpaceRelationId, dst_tblspcoid, GetUserId(),
        ACL_CREATE);
 if (aclresult != ACLCHECK_OK)
  aclcheck_error(aclresult, OBJECT_TABLESPACE,
        tblspcname);

 /*
  * pg_global must never be the default tablespace
 */
 if (dst_tblspcoid == GLOBALTABLESPACE_OID)
  ereport(ERROR,
    (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
     errmsg("pg_global cannot be used as default tablespace")));

 /*
  * No-op if same tablespace
 */
 if (src_tblspcoid == dst_tblspcoid)
 {
  table_close(pgdbrel, NoLock);
  UnlockSharedObjectForSession(DatabaseRelationId, db_id, 0,
          AccessExclusiveLock);
  return;
 }

 /*
  * Check for other backends in the target database.  (Because we hold the
  * database lock, no new ones can start after this.)
  *
  * As in CREATE DATABASE, check this after other error conditions.
 */
 if (CountOtherDBBackends(db_id, ¬herbackends, &npreparedxacts))
  ereport(ERROR,
    (errcode(ERRCODE_OBJECT_IN_USE),
     errmsg("database \"%s\" is being accessed by other users",
      dbname),
     errdetail_busy_db(notherbackends, npreparedxacts)));

 /*
  * Get old and new database paths
 */
 src_dbpath = GetDatabasePath(db_id, src_tblspcoid);
 dst_dbpath = GetDatabasePath(db_id, dst_tblspcoid);

 /*
  * Force a checkpoint before proceeding. This will force all dirty
  * buffers, including those of unlogged tables, out to disk, to ensure
  * source database is up-to-date on disk for the copy.
  * FlushDatabaseBuffers() would suffice for that, but we also want to
  * process any pending unlink requests. Otherwise, the check for existing
  * files in the target directory might fail unnecessarily, not to mention
  * that the copy might fail due to source files getting deleted under it.
  * On Windows, this also ensures that background procs don't hold any open
  * files, which would cause rmdir() to fail.
 */
 RequestCheckpoint(CHECKPOINT_IMMEDIATE | CHECKPOINT_FORCE | CHECKPOINT_WAIT
       | CHECKPOINT_FLUSH_ALL);

 /* Close all smgr fds in all backends. */
 WaitForProcSignalBarrier(EmitProcSignalBarrier(PROCSIGNAL_BARRIER_SMGRRELEASE));

 /*
  * Now drop all buffers holding data of the target database; they should
  * no longer be dirty so DropDatabaseBuffers is safe.
  *
  * It might seem that we could just let these buffers age out of shared
  * buffers naturally, since they should not get referenced anymore.  The
  * problem with that is that if the user later moves the database back to
  * its original tablespace, any still-surviving buffers would appear to
  * contain valid data again --- but they'd be missing any changes made in
  * the database while it was in the new tablespace.  In any case, freeing
  * buffers that should never be used again seems worth the cycles.
  *
  * Note: it'd be sufficient to get rid of buffers matching db_id and
  * src_tblspcoid, but bufmgr.c presently provides no API for that.
 */
 DropDatabaseBuffers(db_id);

 /*
  * Check for existence of files in the target directory, i.e., objects of
  * this database that are already in the target tablespace.  We can't
  * allow the move in such a case, because we would need to change those
  * relations' pg_class.reltablespace entries to zero, and we don't have
  * access to the DB's pg_class to do so.
 */
 dstdir = AllocateDir(dst_dbpath);
 if (dstdir != NULL)
 {
  while ((xlde = ReadDir(dstdir, dst_dbpath)) != NULL)
  {
   if (strcmp(xlde->d_name, ".") == 0 ||
    strcmp(xlde->d_name, "..") == 0)
    continue;

   ereport(ERROR,
     (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
      errmsg("some relations of database \"%s\" are already in tablespace \"%s\"",
       dbname, tblspcname),
      errhint("You must move them back to the database's default tablespace before using this command.")));
  }

  FreeDir(dstdir);

  /*
   * The directory exists but is empty. We must remove it before using
   * the copydir function.
 */
  if (rmdir(dst_dbpath) != 0)
   elog(ERROR, "could not remove directory \"%s\": %m",
     dst_dbpath);
 }

 /*
  * Use an ENSURE block to make sure we remove the debris if the copy fails
  * (eg, due to out-of-disk-space).  This is not a 100% solution, because
  * of the possibility of failure during transaction commit, but it should
  * handle most scenarios.
 */
 fparms.dest_dboid = db_id;
 fparms.dest_tsoid = dst_tblspcoid;
 PG_ENSURE_ERROR_CLEANUP(movedb_failure_callback,
       PointerGetDatum(&fparms));
 {
  Datum  new_record[Natts_pg_database] = {0};
  bool  new_record_nulls[Natts_pg_database] = {0};
  bool  new_record_repl[Natts_pg_database] = {0};

  /*
   * Copy files from the old tablespace to the new one
 */
  copydir(src_dbpath, dst_dbpath, false);

  /*
   * Record the filesystem change in XLOG
 */
  {
   xl_dbase_create_file_copy_rec xlrec;

   xlrec.db_id = db_id;
   xlrec.tablespace_id = dst_tblspcoid;
   xlrec.src_db_id = db_id;
   xlrec.src_tablespace_id = src_tblspcoid;

   XLogBeginInsert();
   XLogRegisterData(&xlrec,
        sizeof(xl_dbase_create_file_copy_rec));

   (void) XLogInsert(RM_DBASE_ID,
         XLOG_DBASE_CREATE_FILE_COPY | XLR_SPECIAL_REL_UPDATE);
  }

  /*
   * Update the database's pg_database tuple
 */
  ScanKeyInit(&scankey,
     Anum_pg_database_datname,
     BTEqualStrategyNumber, F_NAMEEQ,
     CStringGetDatum(dbname));
  sysscan = systable_beginscan(pgdbrel, DatabaseNameIndexId, true,
          NULL, 1, &scankey);
  oldtuple = systable_getnext(sysscan);
  if (!HeapTupleIsValid(oldtuple)) /* shouldn't happen... */
   ereport(ERROR,
     (errcode(ERRCODE_UNDEFINED_DATABASE),
      errmsg("database \"%s\" does not exist", dbname)));
  LockTuple(pgdbrel, &oldtuple->t_self, InplaceUpdateTupleLock);

  new_record[Anum_pg_database_dattablespace - 1] = ObjectIdGetDatum(dst_tblspcoid);
  new_record_repl[Anum_pg_database_dattablespace - 1] = true;

  newtuple = heap_modify_tuple(oldtuple, RelationGetDescr(pgdbrel),
          new_record,
          new_record_nulls, new_record_repl);
  CatalogTupleUpdate(pgdbrel, &oldtuple->t_self, newtuple);
  UnlockTuple(pgdbrel, &oldtuple->t_self, InplaceUpdateTupleLock);

  InvokeObjectPostAlterHook(DatabaseRelationId, db_id, 0);

  systable_endscan(sysscan);

  /*
   * Force another checkpoint here.  As in CREATE DATABASE, this is to
   * ensure that we don't have to replay a committed
   * XLOG_DBASE_CREATE_FILE_COPY operation, which would cause us to lose
   * any unlogged operations done in the new DB tablespace before the
   * next checkpoint.
 */
  RequestCheckpoint(CHECKPOINT_IMMEDIATE | CHECKPOINT_FORCE | CHECKPOINT_WAIT);

  /*
   * Force synchronous commit, thus minimizing the window between
   * copying the database files and committal of the transaction. If we
   * crash before committing, we'll leave an orphaned set of files on
   * disk, which is not fatal but not good either.
 */
  ForceSyncCommit();

  /*
   * Close pg_database, but keep lock till commit.
 */
  table_close(pgdbrel, NoLock);
 }
 PG_END_ENSURE_ERROR_CLEANUP(movedb_failure_callback,
        PointerGetDatum(&fparms));

 /*
  * Commit the transaction so that the pg_database update is committed. If
  * we crash while removing files, the database won't be corrupt, we'll
  * just leave some orphaned files in the old directory.
  *
  * (This is OK because we know we aren't inside a transaction block.)
  *
  * XXX would it be safe/better to do this inside the ensure block? Not
  * convinced it's a good idea; consider elog just after the transaction
  * really commits.
 */
 PopActiveSnapshot();
 CommitTransactionCommand();

 /* Start new transaction for the remaining work; don't need a snapshot */
 StartTransactionCommand();

 /*
  * Remove files from the old tablespace
 */
 if (!rmtree(src_dbpath, true))
  ereport(WARNING,
    (errmsg("some useless files may be left behind in old database directory \"%s\"",
      src_dbpath)));

 /*
  * Record the filesystem change in XLOG
 */
 {
  xl_dbase_drop_rec xlrec;

  xlrec.db_id = db_id;
  xlrec.ntablespaces = 1;

  XLogBeginInsert();
  XLogRegisterData(&xlrec, sizeof(xl_dbase_drop_rec));
  XLogRegisterData(&src_tblspcoid, sizeof(Oid));

  (void) XLogInsert(RM_DBASE_ID,
        XLOG_DBASE_DROP | XLR_SPECIAL_REL_UPDATE);
 }

 /* Now it's safe to release the database lock */
 UnlockSharedObjectForSession(DatabaseRelationId, db_id, 0,
         AccessExclusiveLock);

 pfree(src_dbpath);
 pfree(dst_dbpath);
}

/* Error cleanup callback for movedb */
static void
movedb_failure_callback(int code, Datum arg)
{
 movedb_failure_params *fparms = (movedb_failure_params *) DatumGetPointer(arg);
 char    *dstpath;

 /* Get rid of anything we managed to copy to the target directory */
 dstpath = GetDatabasePath(fparms->dest_dboid, fparms->dest_tsoid);

 (void) rmtree(dstpath, true);

 pfree(dstpath);
}

/*
 * Process options and call dropdb function.
 */
void
DropDatabase(ParseState *pstate, DropdbStmt *stmt)
{
 bool  force = false;
 ListCell   *lc;

 foreach(lc, stmt->options)
 {
  DefElem    *opt = (DefElem *) lfirst(lc);

  if (strcmp(opt->defname, "force") == 0)
   force = true;
  else
   ereport(ERROR,
     (errcode(ERRCODE_SYNTAX_ERROR),
      errmsg("unrecognized %s option \"%s\"",
       "DROP DATABASE", opt->defname),
      parser_errposition(pstate, opt->location)));
 }

 dropdb(stmt->dbname, stmt->missing_ok, force);
}

/*
 * ALTER DATABASE name ...
 */
Oid
AlterDatabase(ParseState *pstate, AlterDatabaseStmt *stmt, bool isTopLevel)
{
 Relation rel;
 Oid   dboid;
 HeapTuple tuple,
    newtuple;
 Form_pg_database datform;
 ScanKeyData scankey;
 SysScanDesc scan;
 ListCell   *option;
 bool  dbistemplate = false;
 bool  dballowconnections = true;
 int   dbconnlimit = DATCONNLIMIT_UNLIMITED;
 DefElem    *distemplate = NULL;
 DefElem    *dallowconnections = NULL;
 DefElem    *dconnlimit = NULL;
 DefElem    *dtablespace = NULL;
 Datum  new_record[Natts_pg_database] = {0};
 bool  new_record_nulls[Natts_pg_database] = {0};
 bool  new_record_repl[Natts_pg_database] = {0};

 /* Extract options from the statement node tree */
 foreach(option, stmt->options)
 {
  DefElem    *defel = (DefElem *) lfirst(option);

  if (strcmp(defel->defname, "is_template") == 0)
  {
   if (distemplate)
    errorConflictingDefElem(defel, pstate);
   distemplate = defel;
  }
  else if (strcmp(defel->defname, "allow_connections") == 0)
  {
   if (dallowconnections)
    errorConflictingDefElem(defel, pstate);
   dallowconnections = defel;
  }
  else if (strcmp(defel->defname, "connection_limit") == 0)
  {
   if (dconnlimit)
    errorConflictingDefElem(defel, pstate);
   dconnlimit = defel;
  }
  else if (strcmp(defel->defname, "tablespace") == 0)
  {
   if (dtablespace)
    errorConflictingDefElem(defel, pstate);
   dtablespace = defel;
  }
  else
   ereport(ERROR,
     (errcode(ERRCODE_SYNTAX_ERROR),
      errmsg("option \"%s\" not recognized", defel->defname),
      parser_errposition(pstate, defel->location)));
 }

 if (dtablespace)
 {
  /*
   * While the SET TABLESPACE syntax doesn't allow any other options,
   * somebody could write "WITH TABLESPACE ...".  Forbid any other
   * options from being specified in that case.
 */
  if (list_length(stmt->options) != 1)
   ereport(ERROR,
     (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
      errmsg("option \"%s\" cannot be specified with other options",
       dtablespace->defname),
      parser_errposition(pstate, dtablespace->location)));
  /* this case isn't allowed within a transaction block */
  PreventInTransactionBlock(isTopLevel, "ALTER DATABASE SET TABLESPACE");
  movedb(stmt->dbname, defGetString(dtablespace));
  return InvalidOid;
 }

 if (distemplate && distemplate->arg)
  dbistemplate = defGetBoolean(distemplate);
 if (dallowconnections && dallowconnections->arg)
  dballowconnections = defGetBoolean(dallowconnections);
 if (dconnlimit && dconnlimit->arg)
 {
  dbconnlimit = defGetInt32(dconnlimit);
  if (dbconnlimit < DATCONNLIMIT_UNLIMITED)
   ereport(ERROR,
     (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
      errmsg("invalid connection limit: %d", dbconnlimit)));
 }

 /*
  * Get the old tuple.  We don't need a lock on the database per se,
  * because we're not going to do anything that would mess up incoming
  * connections.
 */
 rel = table_open(DatabaseRelationId, RowExclusiveLock);
 ScanKeyInit(&scankey,
    Anum_pg_database_datname,
    BTEqualStrategyNumber, F_NAMEEQ,
    CStringGetDatum(stmt->dbname));
 scan = systable_beginscan(rel, DatabaseNameIndexId, true,
         NULL, 1, &scankey);
 tuple = systable_getnext(scan);
 if (!HeapTupleIsValid(tuple))
  ereport(ERROR,
    (errcode(ERRCODE_UNDEFINED_DATABASE),
     errmsg("database \"%s\" does not exist", stmt->dbname)));
 LockTuple(rel, &tuple->t_self, InplaceUpdateTupleLock);

 datform = (Form_pg_database) GETSTRUCT(tuple);
 dboid = datform->oid;

 if (database_is_invalid_form(datform))
 {
  ereport(FATAL,
    errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
    errmsg("cannot alter invalid database \"%s\"", stmt->dbname),
    errhint("Use DROP DATABASE to drop invalid databases."));
 }

 if (!object_ownercheck(DatabaseRelationId, dboid, GetUserId()))
  aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_DATABASE,
        stmt->dbname);

 /*
  * In order to avoid getting locked out and having to go through
  * standalone mode, we refuse to disallow connections to the database
  * we're currently connected to.  Lockout can still happen with concurrent
  * sessions but the likeliness of that is not high enough to worry about.
 */
 if (!dballowconnections && dboid == MyDatabaseId)
  ereport(ERROR,
    (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
     errmsg("cannot disallow connections for current database")));

 /*
  * Build an updated tuple, perusing the information just obtained
 */
 if (distemplate)
 {
  new_record[Anum_pg_database_datistemplate - 1] = BoolGetDatum(dbistemplate);
  new_record_repl[Anum_pg_database_datistemplate - 1] = true;
 }
 if (dallowconnections)
 {
  new_record[Anum_pg_database_datallowconn - 1] = BoolGetDatum(dballowconnections);
  new_record_repl[Anum_pg_database_datallowconn - 1] = true;
 }
 if (dconnlimit)
 {
  new_record[Anum_pg_database_datconnlimit - 1] = Int32GetDatum(dbconnlimit);
  new_record_repl[Anum_pg_database_datconnlimit - 1] = true;
 }

 newtuple = heap_modify_tuple(tuple, RelationGetDescr(rel), new_record,
         new_record_nulls, new_record_repl);
 CatalogTupleUpdate(rel, &tuple->t_self, newtuple);
 UnlockTuple(rel, &tuple->t_self, InplaceUpdateTupleLock);

 InvokeObjectPostAlterHook(DatabaseRelationId, dboid, 0);

 systable_endscan(scan);

 /* Close pg_database, but keep lock till commit */
 table_close(rel, NoLock);

 return dboid;
}


/*
 * ALTER DATABASE name REFRESH COLLATION VERSION
 */
ObjectAddress
AlterDatabaseRefreshColl(AlterDatabaseRefreshCollStmt *stmt)
{
 Relation rel;
 ScanKeyData scankey;
 SysScanDesc scan;
 Oid   db_id;
 HeapTuple tuple;
 Form_pg_database datForm;
 ObjectAddress address;
 Datum  datum;
 bool  isnull;
 char    *oldversion;
 char    *newversion;

 rel = table_open(DatabaseRelationId, RowExclusiveLock);
 ScanKeyInit(&scankey,
    Anum_pg_database_datname,
    BTEqualStrategyNumber, F_NAMEEQ,
    CStringGetDatum(stmt->dbname));
 scan = systable_beginscan(rel, DatabaseNameIndexId, true,
         NULL, 1, &scankey);
 tuple = systable_getnext(scan);
 if (!HeapTupleIsValid(tuple))
  ereport(ERROR,
    (errcode(ERRCODE_UNDEFINED_DATABASE),
     errmsg("database \"%s\" does not exist", stmt->dbname)));

 datForm = (Form_pg_database) GETSTRUCT(tuple);
 db_id = datForm->oid;

 if (!object_ownercheck(DatabaseRelationId, db_id, GetUserId()))
  aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_DATABASE,
        stmt->dbname);
 LockTuple(rel, &tuple->t_self, InplaceUpdateTupleLock);

 datum = heap_getattr(tuple, Anum_pg_database_datcollversion, RelationGetDescr(rel), &isnull);
 oldversion = isnull ? NULL : TextDatumGetCString(datum);

 if (datForm->datlocprovider == COLLPROVIDER_LIBC)
 {
  datum = heap_getattr(tuple, Anum_pg_database_datcollate, RelationGetDescr(rel), &isnull);
  if (isnull)
   elog(ERROR, "unexpected null in pg_database");
 }
 else
 {
  datum = heap_getattr(tuple, Anum_pg_database_datlocale, RelationGetDescr(rel), &isnull);
  if (isnull)
   elog(ERROR, "unexpected null in pg_database");
 }

 newversion = get_collation_actual_version(datForm->datlocprovider,
             TextDatumGetCString(datum));

 /* cannot change from NULL to non-NULL or vice versa */
 if ((!oldversion && newversion) || (oldversion && !newversion))
  elog(ERROR, "invalid collation version change");
 else if (oldversion && newversion && strcmp(newversion, oldversion) != 0)
 {
  bool  nulls[Natts_pg_database] = {0};
  bool  replaces[Natts_pg_database] = {0};
  Datum  values[Natts_pg_database] = {0};
  HeapTuple newtuple;

  ereport(NOTICE,
    (errmsg("changing version from %s to %s",
      oldversion, newversion)));

  values[Anum_pg_database_datcollversion - 1] = CStringGetTextDatum(newversion);
  replaces[Anum_pg_database_datcollversion - 1] = true;

  newtuple = heap_modify_tuple(tuple, RelationGetDescr(rel),
          values, nulls, replaces);
  CatalogTupleUpdate(rel, &tuple->t_self, newtuple);
  heap_freetuple(newtuple);
 }
 else
  ereport(NOTICE,
    (errmsg("version has not changed")));
 UnlockTuple(rel, &tuple->t_self, InplaceUpdateTupleLock);

 InvokeObjectPostAlterHook(DatabaseRelationId, db_id, 0);

 ObjectAddressSet(address, DatabaseRelationId, db_id);

 systable_endscan(scan);

 table_close(rel, NoLock);

 return address;
}


/*
 * ALTER DATABASE name SET ...
 */
Oid
AlterDatabaseSet(AlterDatabaseSetStmt *stmt)
{
 Oid   datid = get_database_oid(stmt->dbname, false);

 /*
  * Obtain a lock on the database and make sure it didn't go away in the
  * meantime.
 */
 shdepLockAndCheckObject(DatabaseRelationId, datid);

 if (!object_ownercheck(DatabaseRelationId, datid, GetUserId()))
  aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_DATABASE,
        stmt->dbname);

 AlterSetting(datid, InvalidOid, stmt->setstmt);

 UnlockSharedObject(DatabaseRelationId, datid, 0, AccessShareLock);

 return datid;
}


/*
 * ALTER DATABASE name OWNER TO newowner
 */
ObjectAddress
AlterDatabaseOwner(const char *dbname, Oid newOwnerId)
{
 Oid   db_id;
 HeapTuple tuple;
 Relation rel;
 ScanKeyData scankey;
 SysScanDesc scan;
 Form_pg_database datForm;
 ObjectAddress address;

 /*
  * Get the old tuple.  We don't need a lock on the database per se,
  * because we're not going to do anything that would mess up incoming
  * connections.
 */
 rel = table_open(DatabaseRelationId, RowExclusiveLock);
 ScanKeyInit(&scankey,
    Anum_pg_database_datname,
    BTEqualStrategyNumber, F_NAMEEQ,
    CStringGetDatum(dbname));
 scan = systable_beginscan(rel, DatabaseNameIndexId, true,
         NULL, 1, &scankey);
 tuple = systable_getnext(scan);
 if (!HeapTupleIsValid(tuple))
  ereport(ERROR,
    (errcode(ERRCODE_UNDEFINED_DATABASE),
     errmsg("database \"%s\" does not exist", dbname)));

 datForm = (Form_pg_database) GETSTRUCT(tuple);
 db_id = datForm->oid;

 /*
  * If the new owner is the same as the existing owner, consider the
  * command to have succeeded.  This is to be consistent with other
  * objects.
 */
 if (datForm->datdba != newOwnerId)
 {
  Datum  repl_val[Natts_pg_database];
  bool  repl_null[Natts_pg_database] = {0};
  bool  repl_repl[Natts_pg_database] = {0};
  Acl     *newAcl;
  Datum  aclDatum;
  bool  isNull;
  HeapTuple newtuple;

  /* Otherwise, must be owner of the existing object */
  if (!object_ownercheck(DatabaseRelationId, db_id, GetUserId()))
   aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_DATABASE,
         dbname);

  /* Must be able to become new owner */
  check_can_set_role(GetUserId(), newOwnerId);

  /*
   * must have createdb rights
   *
   * NOTE: This is different from other alter-owner checks in that the
   * current user is checked for createdb privileges instead of the
   * destination owner.  This is consistent with the CREATE case for
   * databases.  Because superusers will always have this right, we need
   * no special case for them.
 */
  if (!have_createdb_privilege())
   ereport(ERROR,
     (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
      errmsg("permission denied to change owner of database")));

  LockTuple(rel, &tuple->t_self, InplaceUpdateTupleLock);

  repl_repl[Anum_pg_database_datdba - 1] = true;
  repl_val[Anum_pg_database_datdba - 1] = ObjectIdGetDatum(newOwnerId);

  /*
   * Determine the modified ACL for the new owner.  This is only
   * necessary when the ACL is non-null.
 */
  aclDatum = heap_getattr(tuple,
        Anum_pg_database_datacl,
        RelationGetDescr(rel),
        &isNull);
  if (!isNull)
  {
   newAcl = aclnewowner(DatumGetAclP(aclDatum),
         datForm->datdba, newOwnerId);
   repl_repl[Anum_pg_database_datacl - 1] = true;
   repl_val[Anum_pg_database_datacl - 1] = PointerGetDatum(newAcl);
  }

  newtuple = heap_modify_tuple(tuple, RelationGetDescr(rel), repl_val, repl_null, repl_repl);
  CatalogTupleUpdate(rel, &newtuple->t_self, newtuple);
  UnlockTuple(rel, &tuple->t_self, InplaceUpdateTupleLock);

  heap_freetuple(newtuple);

  /* Update owner dependency reference */
  changeDependencyOnOwner(DatabaseRelationId, db_id, newOwnerId);
 }

 InvokeObjectPostAlterHook(DatabaseRelationId, db_id, 0);

 ObjectAddressSet(address, DatabaseRelationId, db_id);

 systable_endscan(scan);

 /* Close pg_database, but keep lock till commit */
 table_close(rel, NoLock);

 return address;
}


Datum
pg_database_collation_actual_version(PG_FUNCTION_ARGS)
{
 Oid   dbid = PG_GETARG_OID(0);
 HeapTuple tp;
 char  datlocprovider;
 Datum  datum;
 char    *version;

 tp = SearchSysCache1(DATABASEOID, ObjectIdGetDatum(dbid));
 if (!HeapTupleIsValid(tp))
  ereport(ERROR,
    (errcode(ERRCODE_UNDEFINED_OBJECT),
     errmsg("database with OID %u does not exist", dbid)));

 datlocprovider = ((Form_pg_database) GETSTRUCT(tp))->datlocprovider;

 if (datlocprovider == COLLPROVIDER_LIBC)
  datum = SysCacheGetAttrNotNull(DATABASEOID, tp, Anum_pg_database_datcollate);
 else
  datum = SysCacheGetAttrNotNull(DATABASEOID, tp, Anum_pg_database_datlocale);

 version = get_collation_actual_version(datlocprovider,
             TextDatumGetCString(datum));

 ReleaseSysCache(tp);

 if (version)
  PG_RETURN_TEXT_P(cstring_to_text(version));
 else
  PG_RETURN_NULL();
}


/*
 * Helper functions
 */

/*
 * Look up info about the database named "name".  If the database exists,
 * obtain the specified lock type on it, fill in any of the remaining
 * parameters that aren't NULL, and return true.  If no such database,
 * return false.
 */
static bool
get_db_info(const char *name, LOCKMODE lockmode,
   Oid *dbIdP, Oid *ownerIdP,
   int *encodingP, bool *dbIsTemplateP, bool *dbAllowConnP, bool *dbHasLoginEvtP,
   TransactionId *dbFrozenXidP, MultiXactId *dbMinMultiP,
   Oid *dbTablespace, char **dbCollate, char **dbCtype, char **dbLocale,
   char **dbIcurules,
   char *dbLocProvider,
   char **dbCollversion)
{
 bool  result = false;
 Relation relation;

 Assert(name);

 /* Caller may wish to grab a better lock on pg_database beforehand... */
 relation = table_open(DatabaseRelationId, AccessShareLock);

 /*
  * Loop covers the rare case where the database is renamed before we can
  * lock it.  We try again just in case we can find a new one of the same
  * name.
 */
 for (;;)
 {
  ScanKeyData scanKey;
  SysScanDesc scan;
  HeapTuple tuple;
  Oid   dbOid;

  /*
   * there's no syscache for database-indexed-by-name, so must do it the
   * hard way
 */
  ScanKeyInit(&scanKey,
     Anum_pg_database_datname,
     BTEqualStrategyNumber, F_NAMEEQ,
     CStringGetDatum(name));

  scan = systable_beginscan(relation, DatabaseNameIndexId, true,
          NULL, 1, &scanKey);

  tuple = systable_getnext(scan);

  if (!HeapTupleIsValid(tuple))
  {
   /* definitely no database of that name */
   systable_endscan(scan);
   break;
  }

  dbOid = ((Form_pg_database) GETSTRUCT(tuple))->oid;

  systable_endscan(scan);

  /*
   * Now that we have a database OID, we can try to lock the DB.
 */
  if (lockmode != NoLock)
   LockSharedObject(DatabaseRelationId, dbOid, 0, lockmode);

  /*
   * And now, re-fetch the tuple by OID.  If it's still there and still
   * the same name, we win; else, drop the lock and loop back to try
   * again.
 */
  tuple = SearchSysCache1(DATABASEOID, ObjectIdGetDatum(dbOid));
  if (HeapTupleIsValid(tuple))
  {
   Form_pg_database dbform = (Form_pg_database) GETSTRUCT(tuple);

   if (strcmp(name, NameStr(dbform->datname)) == 0)
   {
    Datum  datum;
    bool  isnull;

    /* oid of the database */
    if (dbIdP)
     *dbIdP = dbOid;
    /* oid of the owner */
    if (ownerIdP)
     *ownerIdP = dbform->datdba;
    /* character encoding */
    if (encodingP)
     *encodingP = dbform->encoding;
    /* allowed as template? */
    if (dbIsTemplateP)
     *dbIsTemplateP = dbform->datistemplate;
    /* Has on login event trigger? */
    if (dbHasLoginEvtP)
     *dbHasLoginEvtP = dbform->dathasloginevt;
    /* allowing connections? */
    if (dbAllowConnP)
     *dbAllowConnP = dbform->datallowconn;
    /* limit of frozen XIDs */
    if (dbFrozenXidP)
     *dbFrozenXidP = dbform->datfrozenxid;
    /* minimum MultiXactId */
    if (dbMinMultiP)
     *dbMinMultiP = dbform->datminmxid;
    /* default tablespace for this database */
    if (dbTablespace)
     *dbTablespace = dbform->dattablespace;
    /* default locale settings for this database */
    if (dbLocProvider)
     *dbLocProvider = dbform->datlocprovider;
    if (dbCollate)
    {
     datum = SysCacheGetAttrNotNull(DATABASEOID, tuple, Anum_pg_database_datcollate);
     *dbCollate = TextDatumGetCString(datum);
    }
    if (dbCtype)
    {
     datum = SysCacheGetAttrNotNull(DATABASEOID, tuple, Anum_pg_database_datctype);
     *dbCtype = TextDatumGetCString(datum);
    }
    if (dbLocale)
    {
     datum = SysCacheGetAttr(DATABASEOID, tuple, Anum_pg_database_datlocale, &isnull);
     if (isnull)
      *dbLocale = NULL;
     else
      *dbLocale = TextDatumGetCString(datum);
    }
    if (dbIcurules)
    {
     datum = SysCacheGetAttr(DATABASEOID, tuple, Anum_pg_database_daticurules, &isnull);
     if (isnull)
      *dbIcurules = NULL;
     else
      *dbIcurules = TextDatumGetCString(datum);
    }
    if (dbCollversion)
    {
     datum = SysCacheGetAttr(DATABASEOID, tuple, Anum_pg_database_datcollversion, &isnull);
     if (isnull)
      *dbCollversion = NULL;
     else
      *dbCollversion = TextDatumGetCString(datum);
    }
    ReleaseSysCache(tuple);
    result = true;
    break;
   }
   /* can only get here if it was just renamed */
   ReleaseSysCache(tuple);
  }

  if (lockmode != NoLock)
   UnlockSharedObject(DatabaseRelationId, dbOid, 0, lockmode);
 }

 table_close(relation, AccessShareLock);

 return result;
}

/* Check if current user has createdb privileges */
bool
have_createdb_privilege(void)
{
 bool  result = false;
 HeapTuple utup;

 /* Superusers can always do everything */
 if (superuser())
  return true;

 utup = SearchSysCache1(AUTHOID, ObjectIdGetDatum(GetUserId()));
 if (HeapTupleIsValid(utup))
 {
  result = ((Form_pg_authid) GETSTRUCT(utup))->rolcreatedb;
  ReleaseSysCache(utup);
 }
 return result;
}

/*
 * Remove tablespace directories
 *
 * We don't know what tablespaces db_id is using, so iterate through all
 * tablespaces removing <tablespace>/db_id
 */
static void
remove_dbtablespaces(Oid db_id)
{
 Relation rel;
 TableScanDesc scan;
 HeapTuple tuple;
 List    *ltblspc = NIL;
 ListCell   *cell;
 int   ntblspc;
 int   i;
 Oid     *tablespace_ids;

 rel = table_open(TableSpaceRelationId, AccessShareLock);
 scan = table_beginscan_catalog(rel, 0, NULL);
 while ((tuple = heap_getnext(scan, ForwardScanDirection)) != NULL)
 {
  Form_pg_tablespace spcform = (Form_pg_tablespace) GETSTRUCT(tuple);
  Oid   dsttablespace = spcform->oid;
  char    *dstpath;
  struct stat st;

  /* Don't mess with the global tablespace */
  if (dsttablespace == GLOBALTABLESPACE_OID)
   continue;

  dstpath = GetDatabasePath(db_id, dsttablespace);

  if (lstat(dstpath, &st) < 0 || !S_ISDIR(st.st_mode))
  {
   /* Assume we can ignore it */
   pfree(dstpath);
   continue;
  }

  if (!rmtree(dstpath, true))
   ereport(WARNING,
     (errmsg("some useless files may be left behind in old database directory \"%s\"",
       dstpath)));

  ltblspc = lappend_oid(ltblspc, dsttablespace);
  pfree(dstpath);
 }

 ntblspc = list_length(ltblspc);
 if (ntblspc == 0)
 {
  table_endscan(scan);
  table_close(rel, AccessShareLock);
  return;
 }

 tablespace_ids = (Oid *) palloc(ntblspc * sizeof(Oid));
 i = 0;
 foreach(cell, ltblspc)
  tablespace_ids[i++] = lfirst_oid(cell);

 /* Record the filesystem change in XLOG */
 {
  xl_dbase_drop_rec xlrec;

  xlrec.db_id = db_id;
  xlrec.ntablespaces = ntblspc;

  XLogBeginInsert();
  XLogRegisterData(&xlrec, MinSizeOfDbaseDropRec);
  XLogRegisterData(tablespace_ids, ntblspc * sizeof(Oid));

  (void) XLogInsert(RM_DBASE_ID,
        XLOG_DBASE_DROP | XLR_SPECIAL_REL_UPDATE);
 }

 list_free(ltblspc);
 pfree(tablespace_ids);

 table_endscan(scan);
 table_close(rel, AccessShareLock);
}

/*
 * Check for existing files that conflict with a proposed new DB OID;
 * return true if there are any
 *
 * If there were a subdirectory in any tablespace matching the proposed new
 * OID, we'd get a create failure due to the duplicate name ... and then we'd
 * try to remove that already-existing subdirectory during the cleanup in
 * remove_dbtablespaces.  Nuking existing files seems like a bad idea, so
 * instead we make this extra check before settling on the OID of the new
 * database.  This exactly parallels what GetNewRelFileNumber() does for table
 * relfilenumber values.
 */
static bool
check_db_file_conflict(Oid db_id)
{
 bool  result = false;
 Relation rel;
 TableScanDesc scan;
 HeapTuple tuple;

 rel = table_open(TableSpaceRelationId, AccessShareLock);
 scan = table_beginscan_catalog(rel, 0, NULL);
 while ((tuple = heap_getnext(scan, ForwardScanDirection)) != NULL)
 {
  Form_pg_tablespace spcform = (Form_pg_tablespace) GETSTRUCT(tuple);
  Oid   dsttablespace = spcform->oid;
  char    *dstpath;
  struct stat st;

  /* Don't mess with the global tablespace */
  if (dsttablespace == GLOBALTABLESPACE_OID)
   continue;

  dstpath = GetDatabasePath(db_id, dsttablespace);

  if (lstat(dstpath, &st) == 0)
  {
   /* Found a conflicting file (or directory, whatever) */
   pfree(dstpath);
   result = true;
   break;
  }

  pfree(dstpath);
 }

 table_endscan(scan);
 table_close(rel, AccessShareLock);

 return result;
}

/*
 * Issue a suitable errdetail message for a busy database
 */
static int
errdetail_busy_db(int notherbackends, int npreparedxacts)
{
 if (notherbackends > 0 && npreparedxacts > 0)

  /*
   * We don't deal with singular versus plural here, since gettext
   * doesn't support multiple plurals in one string.
 */
  errdetail("There are %d other session(s) and %d prepared transaction(s) using the database.",
      notherbackends, npreparedxacts);
 else if (notherbackends > 0)
  errdetail_plural("There is %d other session using the database.",
       "There are %d other sessions using the database.",
       notherbackends,
       notherbackends);
 else
  errdetail_plural("There is %d prepared transaction using the database.",
       "There are %d prepared transactions using the database.",
       npreparedxacts,
       npreparedxacts);
 return 0;     /* just to keep ereport macro happy */
}

/*
 * get_database_oid - given a database name, look up the OID
 *
 * If missing_ok is false, throw an error if database name not found.  If
 * true, just return InvalidOid.
 */
Oid
get_database_oid(const char *dbname, bool missing_ok)
{
 Relation pg_database;
 ScanKeyData entry[1];
 SysScanDesc scan;
 HeapTuple dbtuple;
 Oid   oid;

 /*
  * There's no syscache for pg_database indexed by name, so we must look
  * the hard way.
 */
 pg_database = table_open(DatabaseRelationId, AccessShareLock);
 ScanKeyInit(&entry[0],
    Anum_pg_database_datname,
    BTEqualStrategyNumber, F_NAMEEQ,
    CStringGetDatum(dbname));
 scan = systable_beginscan(pg_database, DatabaseNameIndexId, true,
         NULL, 1, entry);

 dbtuple = systable_getnext(scan);

 /* We assume that there can be at most one matching tuple */
 if (HeapTupleIsValid(dbtuple))
  oid = ((Form_pg_database) GETSTRUCT(dbtuple))->oid;
 else
  oid = InvalidOid;

 systable_endscan(scan);
 table_close(pg_database, AccessShareLock);

 if (!OidIsValid(oid) && !missing_ok)
  ereport(ERROR,
    (errcode(ERRCODE_UNDEFINED_DATABASE),
     errmsg("database \"%s\" does not exist",
      dbname)));

 return oid;
}


/*
 * get_database_name - given a database OID, look up the name
 *
 * Returns a palloc'd string, or NULL if no such database.
 */
char *
get_database_name(Oid dbid)
{
 HeapTuple dbtuple;
 char    *result;

 dbtuple = SearchSysCache1(DATABASEOID, ObjectIdGetDatum(dbid));
 if (HeapTupleIsValid(dbtuple))
 {
  result = pstrdup(NameStr(((Form_pg_database) GETSTRUCT(dbtuple))->datname));
  ReleaseSysCache(dbtuple);
 }
 else
  result = NULL;

 return result;
}


/*
 * While dropping a database the pg_database row is marked invalid, but the
 * catalog contents still exist. Connections to such a database are not
 * allowed.
 */
bool
database_is_invalid_form(Form_pg_database datform)
{
 return datform->datconnlimit == DATCONNLIMIT_INVALID_DB;
}


/*
 * Convenience wrapper around database_is_invalid_form()
 */
bool
database_is_invalid_oid(Oid dboid)
{
 HeapTuple dbtup;
 Form_pg_database dbform;
 bool  invalid;

 dbtup = SearchSysCache1(DATABASEOID, ObjectIdGetDatum(dboid));
 if (!HeapTupleIsValid(dbtup))
  elog(ERROR, "cache lookup failed for database %u", dboid);
 dbform = (Form_pg_database) GETSTRUCT(dbtup);

 invalid = database_is_invalid_form(dbform);

 ReleaseSysCache(dbtup);

 return invalid;
}


/*
 * recovery_create_dbdir()
 *
 * During recovery, there's a case where we validly need to recover a missing
 * tablespace directory so that recovery can continue.  This happens when
 * recovery wants to create a database but the holding tablespace has been
 * removed before the server stopped.  Since we expect that the directory will
 * be gone before reaching recovery consistency, and we have no knowledge about
 * the tablespace other than its OID here, we create a real directory under
 * pg_tblspc here instead of restoring the symlink.
 *
 * If only_tblspc is true, then the requested directory must be in pg_tblspc/
 */
static void
recovery_create_dbdir(char *path, bool only_tblspc)
{
 struct stat st;

 Assert(RecoveryInProgress());

 if (stat(path, &st) == 0)
  return;

 if (only_tblspc && strstr(path, PG_TBLSPC_DIR_SLASH) == NULL)
  elog(PANIC, "requested to created invalid directory: %s", path);

 if (reachedConsistency && !allow_in_place_tablespaces)
  ereport(PANIC,
    errmsg("missing directory \"%s\"", path));

 elog(reachedConsistency ? WARNING : DEBUG1,
   "creating missing directory: %s", path);

 if (pg_mkdir_p(path, pg_dir_create_mode) != 0)
  ereport(PANIC,
    errmsg("could not create missing directory \"%s\": %m", path));
}


/*
 * DATABASE resource manager's routines
 */
void
dbase_redo(XLogReaderState *record)
{
 uint8  info = XLogRecGetInfo(record) & ~XLR_INFO_MASK;

 /* Backup blocks are not used in dbase records */
 Assert(!XLogRecHasAnyBlockRefs(record));

 if (info == XLOG_DBASE_CREATE_FILE_COPY)
 {
  xl_dbase_create_file_copy_rec *xlrec =
   (xl_dbase_create_file_copy_rec *) XLogRecGetData(record);
  char    *src_path;
  char    *dst_path;
  char    *parent_path;
  struct stat st;

  src_path = GetDatabasePath(xlrec->src_db_id, xlrec->src_tablespace_id);
  dst_path = GetDatabasePath(xlrec->db_id, xlrec->tablespace_id);

  /*
   * Our theory for replaying a CREATE is to forcibly drop the target
   * subdirectory if present, then re-copy the source data. This may be
   * more work than needed, but it is simple to implement.
 */
  if (stat(dst_path, &st) == 0 && S_ISDIR(st.st_mode))
  {
   if (!rmtree(dst_path, true))
    /* If this failed, copydir() below is going to error. */
    ereport(WARNING,
      (errmsg("some useless files may be left behind in old database directory \"%s\"",
        dst_path)));
  }

  /*
   * If the parent of the target path doesn't exist, create it now. This
   * enables us to create the target underneath later.
 */
  parent_path = pstrdup(dst_path);
  get_parent_directory(parent_path);
  if (stat(parent_path, &st) < 0)
  {
   if (errno != ENOENT)
    ereport(FATAL,
      errmsg("could not stat directory \"%s\": %m",
          dst_path));

   /* create the parent directory if needed and valid */
   recovery_create_dbdir(parent_path, true);
  }
  pfree(parent_path);

  /*
   * There's a case where the copy source directory is missing for the
   * same reason above.  Create the empty source directory so that
   * copydir below doesn't fail.  The directory will be dropped soon by
   * recovery.
 */
  if (stat(src_path, &st) < 0 && errno == ENOENT)
   recovery_create_dbdir(src_path, false);

  /*
   * Force dirty buffers out to disk, to ensure source database is
   * up-to-date for the copy.
 */
  FlushDatabaseBuffers(xlrec->src_db_id);

  /* Close all smgr fds in all backends. */
  WaitForProcSignalBarrier(EmitProcSignalBarrier(PROCSIGNAL_BARRIER_SMGRRELEASE));

  /*
   * Copy this subdirectory to the new location
   *
   * We don't need to copy subdirectories
 */
  copydir(src_path, dst_path, false);

  pfree(src_path);
  pfree(dst_path);
 }
 else if (info == XLOG_DBASE_CREATE_WAL_LOG)
 {
  xl_dbase_create_wal_log_rec *xlrec =
   (xl_dbase_create_wal_log_rec *) XLogRecGetData(record);
  char    *dbpath;
  char    *parent_path;

  dbpath = GetDatabasePath(xlrec->db_id, xlrec->tablespace_id);

  /* create the parent directory if needed and valid */
  parent_path = pstrdup(dbpath);
  get_parent_directory(parent_path);
  recovery_create_dbdir(parent_path, true);
  pfree(parent_path);

  /* Create the database directory with the version file. */
  CreateDirAndVersionFile(dbpath, xlrec->db_id, xlrec->tablespace_id,
        true);
  pfree(dbpath);
 }
 else if (info == XLOG_DBASE_DROP)
 {
  xl_dbase_drop_rec *xlrec = (xl_dbase_drop_rec *) XLogRecGetData(record);
  char    *dst_path;
  int   i;

  if (InHotStandby)
  {
   /*
    * Lock database while we resolve conflicts to ensure that
    * InitPostgres() cannot fully re-execute concurrently. This
    * avoids backends re-connecting automatically to same database,
    * which can happen in some cases.
    *
    * This will lock out walsenders trying to connect to db-specific
    * slots for logical decoding too, so it's safe for us to drop
    * slots.
 */
   LockSharedObjectForSession(DatabaseRelationId, xlrec->db_id, 0, AccessExclusiveLock);
   ResolveRecoveryConflictWithDatabase(xlrec->db_id);
  }

  /* Drop any database-specific replication slots */
  ReplicationSlotsDropDBSlots(xlrec->db_id);

  /* Drop pages for this database that are in the shared buffer cache */
  DropDatabaseBuffers(xlrec->db_id);

  /* Also, clean out any fsync requests that might be pending in md.c */
  ForgetDatabaseSyncRequests(xlrec->db_id);

  /* Clean out the xlog relcache too */
  XLogDropDatabase(xlrec->db_id);

  /* Close all smgr fds in all backends. */
  WaitForProcSignalBarrier(EmitProcSignalBarrier(PROCSIGNAL_BARRIER_SMGRRELEASE));

  for (i = 0; i < xlrec->ntablespaces; i++)
  {
   dst_path = GetDatabasePath(xlrec->db_id, xlrec->tablespace_ids[i]);

   /* And remove the physical files */
   if (!rmtree(dst_path, true))
    ereport(WARNING,
      (errmsg("some useless files may be left behind in old database directory \"%s\"",
        dst_path)));
   pfree(dst_path);
  }

  if (InHotStandby)
  {
   /*
    * Release locks prior to commit. XXX There is a race condition
    * here that may allow backends to reconnect, but the window for
    * this is small because the gap between here and commit is mostly
    * fairly small and it is unlikely that people will be dropping
    * databases that we are trying to connect to anyway.
 */
   UnlockSharedObjectForSession(DatabaseRelationId, xlrec->db_id, 0, AccessExclusiveLock);
  }
 }
 else
  elog(PANIC, "dbase_redo: unknown op code %u", info);
}