Function loadMetadata

src/Interpreters/loadMetadata.cpp:207–340

}

LoadTaskPtrs loadMetadata(ContextMutablePtr context, const String & default_database_name, bool async_load_databases)
{
    auto component_guard = Coordination::setCurrentComponent("loadMetadata");
    auto default_db_disk = Context::getGlobalContextInstance()->getDatabaseDisk();

    LoggerPtr log = getLogger("loadMetadata");

    /// There may exist 'force_restore_data' file, which means skip safety threshold
    /// on difference of data parts while initializing tables.
    /// This file is immediately deleted i.e. "one-shot".
    auto force_restore_data_flag_file = fs::path(context->getFlagsPath()) / "force_restore_data";
    bool has_force_restore_data_flag = fs::exists(force_restore_data_flag_file);
    if (has_force_restore_data_flag)
    {
        try
        {
            fs::remove(force_restore_data_flag_file);
        }
        catch (...)
        {
            tryLogCurrentException("Load metadata", "Can't remove force restore file to enable data sanity checks");
        }
    }

    auto metadata_dir_path = DatabaseCatalog::getMetadataDirPath();

    /// Loop over databases.
    std::map<String, fs::path> databases;

    /// Some databases don't have an .sql metadata file.
    std::map<String, fs::path> orphan_directories_and_symlinks;

    for (const auto it = default_db_disk->iterateDirectory(metadata_dir_path); it->isValid(); it->next())
    {
        auto sub_path = fs::path(it->path());
        if (sub_path.filename().empty())
            sub_path = sub_path.parent_path();

        if (default_db_disk->isSymlinkSupported() && default_db_disk->isSymlink(sub_path))
        {
            String db_name = sub_path.filename().string();
            if (!isSystemOrInformationSchema(db_name))
                orphan_directories_and_symlinks.emplace(unescapeForFileName(db_name), fs::path(sub_path.string() + ".sql"));
            continue;
        }

        if (default_db_disk->existsDirectory(sub_path))
            continue;

        if (sub_path.extension() == ".sql")
        {
            String db_name = sub_path.stem();
            orphan_directories_and_symlinks.erase(unescapeForFileName(db_name));
            if (!isSystemOrInformationSchema(db_name))
                databases.emplace(unescapeForFileName(db_name), sub_path);
        }

        /// Temporary fails may be left from previous server runs.