/*-------------------------------------------------------------------------

 *

 * 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_FAST | CHECKPOINT_FORCE |

              CHECKPOINT_WAIT | CHECKPOINT_FLUSH_UNLOGGED);

  /*

   * 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_FAST | 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)