Method createDatabase

src/Interpreters/InterpreterCreateQuery.cpp:200–407

BlockIO InterpreterCreateQuery::createDatabase(ASTCreateQuery & create)
{
    auto component_guard = Coordination::setCurrentComponent("InterpreterCreateQuery::createDatabase");
    String database_name = create.getDatabase();

    auto guard = DatabaseCatalog::instance().getDDLGuard(database_name, "", nullptr);

    /// Database can be created before or it can be created concurrently in another thread, while we were waiting in DDLGuard
    if (DatabaseCatalog::instance().isDatabaseExist(database_name))
    {
        if (create.if_not_exists)
            return {};
        throw Exception(ErrorCodes::DATABASE_ALREADY_EXISTS, "Database {} already exists.", database_name);
    }

    auto db_num_limit = getContext()->getGlobalContext()->getServerSettings()[ServerSetting::max_database_num_to_throw].value;
    if (db_num_limit > 0 && !internal)
    {
        size_t db_count = DatabaseCatalog::instance().getDatabases(GetDatabasesOptions{.with_remote_databases = true}).size();
        std::initializer_list<std::string_view> system_databases =
        {
            DatabaseCatalog::TEMPORARY_DATABASE,
            DatabaseCatalog::SYSTEM_DATABASE,
            DatabaseCatalog::INFORMATION_SCHEMA,
            DatabaseCatalog::INFORMATION_SCHEMA_UPPERCASE,
        };

        for (const auto & system_database : system_databases)
        {
            if (db_count > 0 && DatabaseCatalog::instance().isDatabaseExist(std::string(system_database)))
                --db_count;
        }

        if (db_count >= db_num_limit)
            throw Exception(ErrorCodes::TOO_MANY_DATABASES,
                            "Too many databases. "
                            "The limit (server configuration parameter `max_database_num_to_throw`) is set to {}, the current number of databases is {}",
                            db_num_limit, db_count);
    }

    auto default_db_disk = getContext()->getDatabaseDisk();

    /// Will write file with database metadata, if needed.
    default_db_disk->createDirectories(DatabaseCatalog::getMetadataDirPath());
    auto metadata_file_path = DatabaseCatalog::getMetadataFilePath(database_name);
    auto metadata_tmp_file_path = DatabaseCatalog::getMetadataTmpFilePath(database_name);

    fs::path metadata_path;
    if (!create.storage && create.attach)
    {
        if (!default_db_disk->existsFile(metadata_file_path))
            throw Exception(ErrorCodes::UNKNOWN_DATABASE_ENGINE, "Database engine must be specified for ATTACH DATABASE query");
        /// Short syntax: try read database definition from file
        auto ast = DatabaseOnDisk::parseQueryFromMetadata(nullptr, getContext(), default_db_disk, metadata_file_path);
        create = ast->as<ASTCreateQuery &>();
        if (create.table || !create.storage)
            throw Exception(ErrorCodes::INCORRECT_QUERY, "Metadata file {} contains incorrect CREATE DATABASE query", metadata_file_path.string());
        create.attach = true;