热门标签
热门文章
- 1堆的使用（堆排序和Top-K问题）_top k堆排
- 2mysql创建索引online_InnoDB online DDL与快速索引创建
- 3二分法（又称二分查找算法）
- 4银河麒麟服务器操作系统V10SP2单机和集群部署rocketmq_银河麒麟v10安装mq
- 5ZooKeeper集群环境搭建
- 6Android studio Android SDK下载安装_android studio sdk下载
- 7LintCode编程题：吹气球问题_气球游戏 滑动窗口
- 8【Robotics Library】rlPlanDemo
- 9使用GIT管理文件的步骤（有基础版）_git 文件管理
- 10测试开发 | 相比 Selenium，Web 自动化测试框架 Playwright 有哪些强大的优势？_selenium和playwright库
当前位置: article > 正文
PostgreSQL源码分析——CREATE DATABASE
作者：我家小花儿 | 2024-06-23 04:31:38
赞
踩
PostgreSQL源码分析——CREATE DATABASE
这里我们分析一下在PostgreSQL中创建数据库的源码，在分析源码之前，最好先阅读《PostgreSQL指南内幕探索》的第一章，数据库集簇、数据库和数据表，弄清其空间布局，理解PG中，数据库、表、元组是怎么布局的。通俗一点说，创建一个数据库相当于创建一个数据目录，在这个目录下，存放该数据库中定义的表、索引等对象。在数据库中，存放数据最基础的是表，我们定义了很多表用来存储用户的数据，每个表是一个文件（大于1G的表会分割为多个文件），而表文件内部被划分为固定长度的页，在页中存放元组，这里可以仔细想一下为什么要这么设计？页的大小设置为多少合适？这里涉及到很多方面，比如磁盘，固态硬盘的物理特性，每次读写的最小单元是扇区或块，操作系统为了性能，往往不会按最小单元进行读写，会批量读取，比如8k为一个块进行读取，数据库为了性能会设有缓冲区，用来将最近使用或最常使用的数据缓存起来，这样能避免频繁进行磁盘读取，（PG中使用时钟扫描算法），如果表不划分为页的话，缓存，性能等都不是最合理的方案，实现索引怎么实现的？ 这里可以再读一下《数据库系统内幕》相关章节。
预分析
这我们再分析源码之前预先想一下CREATE DATABASE需要做什么，向pg_database系统表中插入一条数据，数据库名，数据库Oid等信息，创建一个数据库目录（内容从模板数据库copy，目录名为数据库oid）。我们分析一下源码看一下是不是这样。
-- 查看当前数据库 postgres@postgres=# \l List of databases Name | Owner | Encoding | Collate | Ctype | Access privileges -----------+----------+----------+-------------+-------------+----------------------- postgres | postgres | UTF8 | en_US.UTF-8 | en_US.UTF-8 | template0 | postgres | UTF8 | en_US.UTF-8 | en_US.UTF-8 | =c/postgres + | | | | | postgres=CTc/postgres template1 | postgres | UTF8 | en_US.UTF-8 | en_US.UTF-8 | =c/postgres + | | | | | postgres=CTc/postgres (3 rows) -- 查询系统表 postgres@postgres=# select oid,datname,datdba,encoding from pg_database; oid | datname | datdba | encoding -------+-----------+--------+---------- 14204 | postgres | 10 | 6 1 | template1 | 10 | 6 14203 | template0 | 10 | 6 (3 rows) -- 查看数据目录，目录以数据库oid命名 [postgres@slpc base]$ ls 1 14203 14204 postgres@postgres=# \d pg_database; Table "pg_catalog.pg_database" Column | Type | Collation | Nullable | Default ---------------+-----------+-----------+----------+--------- oid | oid | | not null | datname | name | | not null | datdba | oid | | not null | encoding | integer | | not null | datcollate | name | | not null | datctype | name | | not null | datistemplate | boolean | | not null | datallowconn | boolean | | not null | datconnlimit | integer | | not null | datlastsysoid | oid | | not null | datfrozenxid | xid | | not null | datminmxid | xid | | not null | dattablespace | oid | | not null | datacl | aclitem[] | | | Indexes: "pg_database_datname_index" UNIQUE, btree (datname), tablespace "pg_global" "pg_database_oid_index" UNIQUE, btree (oid), tablespace "pg_global" Tablespace: "pg_global"
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
CREATE DATABASE语法使用如下：
CREATE DATABASE name [ WITH ] [ OWNER [=] user_name ] [ TEMPLATE [=] template ] [ ENCODING [=] encoding ] [ STRATEGY [=] strategy ] ] [ LOCALE [=] locale ] [ LC_COLLATE [=] lc_collate ] [ LC_CTYPE [=] lc_ctype ] [ ICU_LOCALE [=] icu_locale ] [ LOCALE_PROVIDER [=] locale_provider ] [ COLLATION_VERSION = collation_version ] [ TABLESPACE [=] tablespace_name ] [ ALLOW_CONNECTIONS [=] allowconn ] [ CONNECTION LIMIT [=] connlimit ] [ IS_TEMPLATE [=] istemplate ] [ OID [=] oid ]
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
pg_database系统表参考文档：https://www.postgresql.org/docs/current/catalog-pg-database.html
语法参考文档：https://www.postgresql.org/docs/current/sql-createdatabase.html
源码分析
其语法相关可参考：
解析层
关键定义：
/* ----------------------
* Createdb Statement
* ---------------------- */
typedef struct CreatedbStmt
{
NodeTag type;
char *dbname; /* name of database to create */
List *options; /* List of DefElem nodes */
} CreatedbStmt;
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
语法定义(gram.y)：
CreatedbStmt:
CREATE DATABASE name opt_with createdb_opt_list
{
CreatedbStmt *n = makeNode(CreatedbStmt);
n->dbname = $3; // 最重要的是告诉数据库，创建的数据库名是什么
n->options = $5; // 创建数据库的一些选项，比如encoding, tablespace, location等等
$$ = (Node *) n;
}
;
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
执行层
CREATE DATABASE属于非优化语句，跳过优化器，直接进入执行器执行对应的操作。
exec_simple_query
--> pg_parse_query
--> raw_parser
--> pg_analyze_and_rewrite
--> parse_analyze
--> transformStmt
--> pg_plan_queries
--> PortalStart
--> PortalRun
--> PortalRunMulti
--> ProcessUtility
--> standard_ProcessUtility
--> createdb // 执行创建数据库操作
--> PortalDrop
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
下面是createdb函数的具体代码，其中一点需要注意，PG采用从模板数据库复制的方式来创建新的数据库，没有指定的话，默认模板是template1数据库。
创建数据库代码如下：
/* 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_iculocale = NULL; char src_locprovider = '\0'; char *src_collversion = NULL; bool src_istemplate; 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]; bool new_record_nulls[Natts_pg_database]; Oid dboid = InvalidOid; Oid datdba; ListCell *option; DefElem *dtablespacename = NULL; DefElem *downer = NULL; DefElem *dtemplate = NULL; DefElem *dencoding = NULL; DefElem *dlocale = NULL; DefElem *dcollate = NULL; DefElem *dctype = NULL; DefElem *diculocale = NULL; DefElem *dlocprovider = NULL; DefElem *distemplate = NULL; DefElem *dallowconnections = NULL; DefElem *dconnlimit = NULL; DefElem *dcollversion = NULL; DefElem *dstrategy = NULL; char *dbname = stmt->dbname; char *dbowner = NULL; const char *dbtemplate = NULL; char *dbcollate = NULL; char *dbctype = NULL; char *dbiculocale = NULL; char dblocprovider = '\0'; char *canonname; int encoding = -1; bool dbistemplate = false; bool dballowconnections = true; int dbconnlimit = -1; 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 (dtablespacename) errorConflictingDefElem(defel, pstate); dtablespacename = defel; } else if (strcmp(defel->defname, "owner") == 0) { if (downer) errorConflictingDefElem(defel, pstate); downer = defel; } else if (strcmp(defel->defname, "template") == 0) { if (dtemplate) errorConflictingDefElem(defel, pstate); dtemplate = defel; } else if (strcmp(defel->defname, "encoding") == 0) { if (dencoding) errorConflictingDefElem(defel, pstate); dencoding = defel; } else if (strcmp(defel->defname, "locale") == 0) { if (dlocale) errorConflictingDefElem(defel, pstate); dlocale = defel; } else if (strcmp(defel->defname, "lc_collate") == 0) { if (dcollate) errorConflictingDefElem(defel, pstate); dcollate = defel; } else if (strcmp(defel->defname, "lc_ctype") == 0) { if (dctype) errorConflictingDefElem(defel, pstate); dctype = defel; } else if (strcmp(defel->defname, "icu_locale") == 0) { if (diculocale) errorConflictingDefElem(defel, pstate); diculocale = defel; } else if (strcmp(defel->defname, "locale_provider") == 0) { if (dlocprovider) errorConflictingDefElem(defel, pstate); dlocprovider = defel; } else 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, "collation_version") == 0) { if (dcollversion) errorConflictingDefElem(defel, pstate); dcollversion = 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 (dstrategy) errorConflictingDefElem(defel, pstate); dstrategy = defel; } else ereport(ERROR, (errcode(ERRCODE_SYNTAX_ERROR), errmsg("option \"%s\" not recognized", defel->defname), parser_errposition(pstate, defel->location))); } if (downer && downer->arg) dbowner = defGetString(downer); if (dtemplate && dtemplate->arg) dbtemplate = defGetString(dtemplate); if (dencoding && dencoding->arg) { const char *encoding_name; if (IsA(dencoding->arg, Integer)) { encoding = defGetInt32(dencoding); 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, dencoding->location))); } else { encoding_name = defGetString(dencoding); 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, dencoding->location))); } } if (dlocale && dlocale->arg) { dbcollate = defGetString(dlocale); dbctype = defGetString(dlocale); } if (dcollate && dcollate->arg) dbcollate = defGetString(dcollate); if (dctype && dctype->arg) dbctype = defGetString(dctype); if (diculocale && diculocale->arg) dbiculocale = defGetString(diculocale); if (dlocprovider && dlocprovider->arg) { char *locproviderstr = defGetString(dlocprovider); 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 (distemplate && distemplate->arg) dbistemplate = defGetBoolean(distemplate); if (dallowconnections && dallowconnections->arg) dballowconnections = defGetBoolean(dallowconnections); if (dconnlimit && dconnlimit->arg) { dbconnlimit = defGetInt32(dconnlimit); if (dbconnlimit < -1) ereport(ERROR, (errcode(ERRCODE_INVALID_PARAMETER_VALUE), errmsg("invalid connection limit: %d", dbconnlimit))); } if (dcollversion) dbcollversion = defGetString(dcollversion); /* 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_is_member_of_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_frozenxid, &src_minmxid, &src_deftablespace, &src_collate, &src_ctype, &src_iculocale, &src_locprovider, &src_collversion)) ereport(ERROR, (errcode(ERRCODE_UNDEFINED_DATABASE), errmsg("template database \"%s\" does not exist", dbtemplate))); /* * Permission check: to copy a DB that's not marked datistemplate, you * must be superuser or the owner thereof. */ if (!src_istemplate) { if (!pg_database_ownercheck(src_dboid, GetUserId())) ereport(ERROR, (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE), errmsg("permission denied to copy database \"%s\"", dbtemplate))); } /* Validate the database creation strategy. */ if (dstrategy && dstrategy->arg) { char *strategy; strategy = defGetString(dstrategy); if (strcmp(strategy, "wal_log") == 0) dbstrategy = CREATEDB_WAL_LOG; else if (strcmp(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 (dbiculocale == NULL && dblocprovider == COLLPROVIDER_ICU) dbiculocale = src_iculocale; /* 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)) ereport(ERROR, (errcode(ERRCODE_WRONG_OBJECT_TYPE), errmsg("invalid locale name: \"%s\"", dbcollate))); dbcollate = canonname; if (!check_locale(LC_CTYPE, dbctype, &canonname)) ereport(ERROR, (errcode(ERRCODE_WRONG_OBJECT_TYPE), errmsg("invalid locale name: \"%s\"", dbctype))); dbctype = canonname; check_encoding_locale_matches(encoding, dbcollate, dbctype); 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 (!dbiculocale) ereport(ERROR, (errcode(ERRCODE_INVALID_PARAMETER_VALUE), errmsg("ICU locale must be specified"))); check_icu_locale(dbiculocale); } else { if (dbiculocale) ereport(ERROR, (errcode(ERRCODE_INVALID_OBJECT_DEFINITION), errmsg("ICU locale cannot be specified unless locale provider is ICU"))); } /* * 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) { Assert(dbiculocale); Assert(src_iculocale); if (strcmp(dbiculocale, src_iculocale) != 0) ereport(ERROR, (errcode(ERRCODE_INVALID_PARAMETER_VALUE), errmsg("new ICU locale (%s) is incompatible with the ICU locale of the template database (%s)", dbiculocale, src_iculocale), errhint("Use the same ICU locale 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 && !dcollversion) { char *actual_versionstr; actual_versionstr = get_collation_actual_version(dblocprovider, dblocprovider == COLLPROVIDER_ICU ? dbiculocale : dbcollate); 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) dbcollversion = get_collation_actual_version(dblocprovider, dblocprovider == COLLPROVIDER_ICU ? dbiculocale : dbcollate); /* Resolve default tablespace for new database */ if (dtablespacename && dtablespacename->arg) { char *tablespacename; AclResult aclresult; tablespacename = defGetString(dtablespacename); dst_deftablespace = get_tablespace_oid(tablespacename, false); /* check permissions */ aclresult = pg_tablespace_aclcheck(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_ICU && dbiculocale) || (dblocprovider != COLLPROVIDER_ICU && !dbiculocale)); /* Form tuple */ MemSet(new_record, 0, sizeof(new_record)); MemSet(new_record_nulls, false, sizeof(new_record_nulls)); 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_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 (dbiculocale) new_record[Anum_pg_database_daticulocale - 1] = CStringGetTextDatum(dbiculocale); else new_record_nulls[Anum_pg_database_daticulocale - 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 a * lock 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; }
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
- 58
- 59
- 60
- 61
- 62
- 63
- 64
- 65
- 66
- 67
- 68
- 69
- 70
- 71
- 72
- 73
- 74
- 75
- 76
- 77
- 78
- 79
- 80
- 81
- 82
- 83
- 84
- 85
- 86
- 87
- 88
- 89
- 90
- 91
- 92
- 93
- 94
- 95
- 96
- 97
- 98
- 99
- 100
- 101
- 102
- 103
- 104
- 105
- 106
- 107
- 108
- 109
- 110
- 111
- 112
- 113
- 114
- 115
- 116
- 117
- 118
- 119
- 120
- 121
- 122
- 123
- 124
- 125
- 126
- 127
- 128
- 129
- 130
- 131
- 132
- 133
- 134
- 135
- 136
- 137
- 138
- 139
- 140
- 141
- 142
- 143
- 144
- 145
- 146
- 147
- 148
- 149
- 150
- 151
- 152
- 153
- 154
- 155
- 156
- 157
- 158
- 159
- 160
- 161
- 162
- 163
- 164
- 165
- 166
- 167
- 168
- 169
- 170
- 171
- 172
- 173
- 174
- 175
- 176
- 177
- 178
- 179
- 180
- 181
- 182
- 183
- 184
- 185
- 186
- 187
- 188
- 189
- 190
- 191
- 192
- 193
- 194
- 195
- 196
- 197
- 198
- 199
- 200
- 201
- 202
- 203
- 204
- 205
- 206
- 207
- 208
- 209
- 210
- 211
- 212
- 213
- 214
- 215
- 216
- 217
- 218
- 219
- 220
- 221
- 222
- 223
- 224
- 225
- 226
- 227
- 228
- 229
- 230
- 231
- 232
- 233
- 234
- 235
- 236
- 237
- 238
- 239
- 240
- 241
- 242
- 243
- 244
- 245
- 246
- 247
- 248
- 249
- 250
- 251
- 252
- 253
- 254
- 255
- 256
- 257
- 258
- 259
- 260
- 261
- 262
- 263
- 264
- 265
- 266
- 267
- 268
- 269
- 270
- 271
- 272
- 273
- 274
- 275
- 276
- 277
- 278
- 279
- 280
- 281
- 282
- 283
- 284
- 285
- 286
- 287
- 288
- 289
- 290
- 291
- 292
- 293
- 294
- 295
- 296
- 297
- 298
- 299
- 300
- 301
- 302
- 303
- 304
- 305
- 306
- 307
- 308
- 309
- 310
- 311
- 312
- 313
- 314
- 315
- 316
- 317
- 318
- 319
- 320
- 321
- 322
- 323
- 324
- 325
- 326
- 327
- 328
- 329
- 330
- 331
- 332
- 333
- 334
- 335
- 336
- 337
- 338
- 339
- 340
- 341
- 342
- 343
- 344
- 345
- 346
- 347
- 348
- 349
- 350
- 351
- 352
- 353
- 354
- 355
- 356
- 357
- 358
- 359
- 360
- 361
- 362
- 363
- 364
- 365
- 366
- 367
- 368
- 369
- 370
- 371
- 372
- 373
- 374
- 375
- 376
- 377
- 378
- 379
- 380
- 381
- 382
- 383
- 384
- 385
- 386
- 387
- 388
- 389
- 390
- 391
- 392
- 393
- 394
- 395
- 396
- 397
- 398
- 399
- 400
- 401
- 402
- 403
- 404
- 405
- 406
- 407
- 408
- 409
- 410
- 411
- 412
- 413
- 414
- 415
- 416
- 417
- 418
- 419
- 420
- 421
- 422
- 423
- 424
- 425
- 426
- 427
- 428
- 429
- 430
- 431
- 432
- 433
- 434
- 435
- 436
- 437
- 438
- 439
- 440
- 441
- 442
- 443
- 444
- 445
- 446
- 447
- 448
- 449
- 450
- 451
- 452
- 453
- 454
- 455
- 456
- 457
- 458
- 459
- 460
- 461
- 462
- 463
- 464
- 465
- 466
- 467
- 468
- 469
- 470
- 471
- 472
- 473
- 474
- 475
- 476
- 477
- 478
- 479
- 480
- 481
- 482
- 483
- 484
- 485
- 486
- 487
- 488
- 489
- 490
- 491
- 492
- 493
- 494
- 495
- 496
- 497
- 498
- 499
- 500
- 501
- 502
- 503
- 504
- 505
- 506
- 507
- 508
- 509
- 510
- 511
- 512
- 513
- 514
- 515
- 516
- 517
- 518
- 519
- 520
- 521
- 522
- 523
- 524
- 525
- 526
- 527
- 528
- 529
- 530
- 531
- 532
- 533
- 534
- 535
- 536
- 537
- 538
- 539
- 540
- 541
- 542
- 543
- 544
- 545
- 546
- 547
- 548
- 549
- 550
- 551
- 552
- 553
- 554
- 555
- 556
- 557
- 558
- 559
- 560
- 561
- 562
- 563
- 564
- 565
- 566
- 567
- 568
- 569
- 570
- 571
- 572
- 573
- 574
- 575
- 576
- 577
- 578
- 579
- 580
- 581
- 582
- 583
- 584
- 585
- 586
- 587
- 588
- 589
- 590
- 591
- 592
- 593
- 594
- 595
- 596
- 597
- 598
- 599
- 600
- 601
- 602
- 603
- 604
- 605
- 606
- 607
- 608
- 609
- 610
- 611
- 612
- 613
- 614
- 615
- 616
- 617
- 618
- 619
- 620
- 621
- 622
- 623
- 624
- 625
- 626
- 627
- 628
- 629
- 630
- 631
- 632
- 633
- 634
- 635
- 636
- 637
- 638
- 639
- 640
- 641
- 642
- 643
- 644
- 645
- 646
- 647
- 648
- 649
- 650
- 651
- 652
- 653
- 654
- 655
- 656
- 657
- 658
- 659
- 660
- 661
- 662
- 663
- 664
- 665
- 666
- 667
- 668
- 669
- 670
- 671
- 672
- 673
- 674
- 675
- 676
- 677
- 678
- 679
- 680
- 681
- 682
- 683
- 684
- 685
- 686
- 687
- 688
- 689
- 690
- 691
- 692
- 693
- 694
- 695
- 696
- 697
- 698
- 699
- 700
- 701
- 702
- 703
- 704
- 705
- 706
- 707
- 708
- 709
- 710
- 711
- 712
- 713
- 714
- 715
- 716
- 717
- 718
- 719
- 720
- 721
- 722
- 723
- 724
- 725
- 726
- 727
- 728
- 729
- 730
- 731
- 732
- 733
- 734
- 735
- 736
- 737
- 738
- 739
- 740
声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/我家小花儿/article/detail/748651
推荐阅读
相关标签
