mirror of https://github.com/ByConity/ByConity
Dictionaries can be loaded in parallel.
A bad dictionary cannot block all the loading anymore. Implemented really lazy loading of external dictionaries. Provided more detailed information about the loading of each dictionary to make diagnostics easier.
This commit is contained in:
Vitaly Baranov
parent
1ecc430335
commit
b4384ce2a9
|
|
@ -33,6 +33,7 @@
|
|||
#include <IO/UseSSL.h>
|
||||
#include <Interpreters/AsynchronousMetrics.h>
|
||||
#include <Interpreters/DDLWorker.h>
|
||||
#include <Interpreters/ExternalDictionaries.h>
|
||||
#include <Interpreters/ProcessList.h>
|
||||
#include <Interpreters/loadMetadata.h>
|
||||
#include <Interpreters/DNSCacheUpdater.h>
|
||||
|
|
@ -832,7 +833,7 @@ int Server::main(const std::vector<std::string> & /*args*/)
|
|||
if (!config().getBool("dictionaries_lazy_load", true))
|
||||
{
|
||||
global_context->tryCreateEmbeddedDictionaries();
|
||||
global_context->tryCreateExternalDictionaries();
|
||||
global_context->getExternalDictionaries().enableAlwaysLoadEverything(true);
|
||||
}
|
||||
}
|
||||
catch (...)
|
||||
|
|
|
|||
|
|
@ -0,0 +1,70 @@
|
|||
#include <Common/Config/AbstractConfigurationComparison.h>
|
||||
|
||||
#include <unordered_set>
|
||||
#include <common/StringRef.h>
|
||||
#include <Poco/Util/AbstractConfiguration.h>
|
||||
|
||||
|
||||
namespace DB
|
||||
{
|
||||
namespace
|
||||
{
|
||||
String concatKeyAndSubKey(const String & key, const String & subkey)
|
||||
{
|
||||
// Copied from Poco::Util::ConfigurationView::translateKey():
|
||||
String result = key;
|
||||
if (!result.empty() && !subkey.empty() && subkey[0] != '[')
|
||||
result += '.';
|
||||
result += subkey;
|
||||
return result;
|
||||
};
|
||||
}
|
||||
|
||||
|
||||
bool isSameConfiguration(const Poco::Util::AbstractConfiguration & left, const Poco::Util::AbstractConfiguration & right)
|
||||
{
|
||||
return isSameConfiguration(left, String(), right, String());
|
||||
}
|
||||
|
||||
|
||||
bool isSameConfiguration(const Poco::Util::AbstractConfiguration & left, const String & left_key,
|
||||
const Poco::Util::AbstractConfiguration & right, const String & right_key)
|
||||
{
|
||||
if (&left == &right && left_key == right_key)
|
||||
return true;
|
||||
|
||||
bool has_property = left.hasProperty(left_key);
|
||||
if (has_property != right.hasProperty(right_key))
|
||||
return false;
|
||||
if (has_property)
|
||||
{
|
||||
/// The left and right configurations contains values so we can compare them.
|
||||
if (left.getRawString(left_key) != right.getRawString(right_key))
|
||||
return false;
|
||||
}
|
||||
|
||||
/// Get the subkeys of the left and right configurations.
|
||||
Poco::Util::AbstractConfiguration::Keys subkeys;
|
||||
left.keys(left_key, subkeys);
|
||||
|
||||
{
|
||||
/// Check that the right configuration has the same set of subkeys as the left configuration.
|
||||
Poco::Util::AbstractConfiguration::Keys right_subkeys;
|
||||
right.keys(right_key, right_subkeys);
|
||||
std::unordered_set<StringRef> left_subkeys{subkeys.begin(), subkeys.end()};
|
||||
if ((left_subkeys.size() != right_subkeys.size()) || (left_subkeys.size() != subkeys.size()))
|
||||
return false;
|
||||
for (const auto & right_subkey : right_subkeys)
|
||||
if (!left_subkeys.count(right_subkey))
|
||||
return false;
|
||||
}
|
||||
|
||||
/// Go through all the subkeys and compare corresponding parts of the configurations.
|
||||
for (const auto & subkey : subkeys)
|
||||
if (!isSameConfiguration(left, concatKeyAndSubKey(left_key, subkey), right, concatKeyAndSubKey(right_key, subkey)))
|
||||
return false;
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -0,0 +1,29 @@
|
|||
#pragma once
|
||||
|
||||
#include <Core/Types.h>
|
||||
|
||||
namespace Poco::Util
|
||||
{
|
||||
class AbstractConfiguration;
|
||||
}
|
||||
|
||||
namespace DB
|
||||
{
|
||||
/// Returns true if two configurations contains the same keys and values.
|
||||
bool isSameConfiguration(const Poco::Util::AbstractConfiguration & left,
|
||||
const Poco::Util::AbstractConfiguration & right);
|
||||
|
||||
/// Returns true if specified subviews of the two configurations contains the same keys and values.
|
||||
bool isSameConfiguration(const Poco::Util::AbstractConfiguration & left, const String & left_key,
|
||||
const Poco::Util::AbstractConfiguration & right, const String & right_key);
|
||||
|
||||
inline bool operator==(const Poco::Util::AbstractConfiguration & left, const Poco::Util::AbstractConfiguration & right)
|
||||
{
|
||||
return isSameConfiguration(left, right);
|
||||
}
|
||||
|
||||
inline bool operator!=(const Poco::Util::AbstractConfiguration & left, const Poco::Util::AbstractConfiguration & right)
|
||||
{
|
||||
return !isSameConfiguration(left, right);
|
||||
}
|
||||
}
|
||||
|
|
@ -31,24 +31,29 @@ void DatabaseDictionary::loadTables(Context &, ThreadPool *, bool)
|
|||
{
|
||||
}
|
||||
|
||||
Tables DatabaseDictionary::listTables(const Context & context)
|
||||
Tables DatabaseDictionary::listTables(const Context & context, const FilterByNameFunction & filter_by_name)
|
||||
{
|
||||
auto objects_map = context.getExternalDictionaries().getObjectsMap();
|
||||
const auto & dictionaries = objects_map.get();
|
||||
|
||||
Tables tables;
|
||||
for (const auto & pair : dictionaries)
|
||||
ExternalLoader::Loadables loadables;
|
||||
if (filter_by_name)
|
||||
{
|
||||
auto dict_ptr = std::static_pointer_cast<IDictionaryBase>(pair.second.loadable);
|
||||
if (dict_ptr)
|
||||
{
|
||||
const DictionaryStructure & dictionary_structure = dict_ptr->getStructure();
|
||||
auto columns = StorageDictionary::getNamesAndTypes(dictionary_structure);
|
||||
const std::string & dict_name = pair.first;
|
||||
tables[dict_name] = StorageDictionary::create(dict_name, ColumnsDescription{columns}, context, true, dict_name);
|
||||
}
|
||||
/// If `filter_by_name` is set, we iterate through all dictionaries with such names. That's why we need to load all of them.
|
||||
loadables = context.getExternalDictionaries().loadAndGet(filter_by_name);
|
||||
}
|
||||
else
|
||||
{
|
||||
/// If `filter_by_name` isn't set, we iterate through only already loaded dictionaries. We don't try to load all dictionaries in this case.
|
||||
loadables = context.getExternalDictionaries().getCurrentlyLoadedObjects();
|
||||
}
|
||||
|
||||
for (const auto & loadable : loadables)
|
||||
{
|
||||
auto dict_ptr = std::static_pointer_cast<const IDictionaryBase>(loadable);
|
||||
auto dict_name = dict_ptr->getName();
|
||||
const DictionaryStructure & dictionary_structure = dict_ptr->getStructure();
|
||||
auto columns = StorageDictionary::getNamesAndTypes(dictionary_structure);
|
||||
tables[dict_name] = StorageDictionary::create(dict_name, ColumnsDescription{columns}, context, true, dict_name);
|
||||
}
|
||||
return tables;
|
||||
}
|
||||
|
||||
|
|
@ -56,9 +61,7 @@ bool DatabaseDictionary::isTableExist(
|
|||
const Context & context,
|
||||
const String & table_name) const
|
||||
{
|
||||
auto objects_map = context.getExternalDictionaries().getObjectsMap();
|
||||
const auto & dictionaries = objects_map.get();
|
||||
return dictionaries.count(table_name);
|
||||
return context.getExternalDictionaries().getCurrentStatus(table_name) != ExternalLoader::Status::NOT_EXIST;
|
||||
}
|
||||
|
||||
StoragePtr DatabaseDictionary::tryGetTable(
|
||||
|
|
@ -76,19 +79,14 @@ StoragePtr DatabaseDictionary::tryGetTable(
|
|||
return {};
|
||||
}
|
||||
|
||||
DatabaseIteratorPtr DatabaseDictionary::getIterator(const Context & context)
|
||||
DatabaseIteratorPtr DatabaseDictionary::getIterator(const Context & context, const FilterByNameFunction & filter_by_name)
|
||||
{
|
||||
return std::make_unique<DatabaseSnapshotIterator>(listTables(context));
|
||||
return std::make_unique<DatabaseSnapshotIterator>(listTables(context, filter_by_name));
|
||||
}
|
||||
|
||||
bool DatabaseDictionary::empty(const Context & context) const
|
||||
{
|
||||
auto objects_map = context.getExternalDictionaries().getObjectsMap();
|
||||
const auto & dictionaries = objects_map.get();
|
||||
for (const auto & pair : dictionaries)
|
||||
if (pair.second.loadable)
|
||||
return false;
|
||||
return true;
|
||||
return context.getExternalDictionaries().getNumberOfNames() == 0;
|
||||
}
|
||||
|
||||
StoragePtr DatabaseDictionary::detachTable(const String & /*table_name*/)
|
||||
|
|
|
|||
|
|
@ -44,7 +44,7 @@ public:
|
|||
const Context & context,
|
||||
const String & table_name) const override;
|
||||
|
||||
DatabaseIteratorPtr getIterator(const Context & context) override;
|
||||
DatabaseIteratorPtr getIterator(const Context & context, const FilterByNameFunction & filter_by_table_name = {}) override;
|
||||
|
||||
bool empty(const Context & context) const override;
|
||||
|
||||
|
|
@ -96,7 +96,7 @@ private:
|
|||
|
||||
Poco::Logger * log;
|
||||
|
||||
Tables listTables(const Context & context);
|
||||
Tables listTables(const Context & context, const FilterByNameFunction & filter_by_name);
|
||||
ASTPtr getCreateTableQueryImpl(const Context & context, const String & table_name, bool throw_on_error) const;
|
||||
};
|
||||
|
||||
|
|
|
|||
|
|
@ -107,10 +107,16 @@ StoragePtr DatabaseWithOwnTablesBase::tryGetTable(
|
|||
return it->second;
|
||||
}
|
||||
|
||||
DatabaseIteratorPtr DatabaseWithOwnTablesBase::getIterator(const Context & /*context*/)
|
||||
DatabaseIteratorPtr DatabaseWithOwnTablesBase::getIterator(const Context & /*context*/, const FilterByNameFunction & filter_by_table_name)
|
||||
{
|
||||
std::lock_guard lock(mutex);
|
||||
return std::make_unique<DatabaseSnapshotIterator>(tables);
|
||||
if (!filter_by_table_name)
|
||||
return std::make_unique<DatabaseSnapshotIterator>(tables);
|
||||
Tables filtered_tables;
|
||||
for (const auto & [table_name, storage] : tables)
|
||||
if (filter_by_table_name(table_name))
|
||||
filtered_tables.emplace(table_name, storage);
|
||||
return std::make_unique<DatabaseSnapshotIterator>(std::move(filtered_tables));
|
||||
}
|
||||
|
||||
bool DatabaseWithOwnTablesBase::empty(const Context & /*context*/) const
|
||||
|
|
|
|||
|
|
@ -89,7 +89,7 @@ public:
|
|||
|
||||
StoragePtr detachTable(const String & table_name) override;
|
||||
|
||||
DatabaseIteratorPtr getIterator(const Context & context) override;
|
||||
DatabaseIteratorPtr getIterator(const Context & context, const FilterByNameFunction & filter_by_table_name = {}) override;
|
||||
|
||||
void shutdown() override;
|
||||
|
||||
|
|
|
|||
|
|
@ -73,9 +73,11 @@ public:
|
|||
const Context & context,
|
||||
const String & name) const = 0;
|
||||
|
||||
using FilterByNameFunction = std::function<bool(const String &)>;
|
||||
|
||||
/// Get an iterator that allows you to pass through all the tables.
|
||||
/// It is possible to have "hidden" tables that are not visible when passing through, but are visible if you get them by name using the functions above.
|
||||
virtual DatabaseIteratorPtr getIterator(const Context & context) = 0;
|
||||
virtual DatabaseIteratorPtr getIterator(const Context & context, const FilterByNameFunction & filter_by_table_name = {}) = 0;
|
||||
|
||||
/// Is the database empty.
|
||||
virtual bool empty(const Context & context) const = 0;
|
||||
|
|
|
|||
|
|
@ -51,9 +51,9 @@ public:
|
|||
|
||||
bool isCached() const override { return true; }
|
||||
|
||||
std::unique_ptr<IExternalLoadable> clone() const override
|
||||
std::shared_ptr<const IExternalLoadable> clone() const override
|
||||
{
|
||||
return std::make_unique<CacheDictionary>(name, dict_struct, source_ptr->clone(), dict_lifetime, size);
|
||||
return std::make_shared<CacheDictionary>(name, dict_struct, source_ptr->clone(), dict_lifetime, size);
|
||||
}
|
||||
|
||||
const IDictionarySource * getSource() const override { return source_ptr.get(); }
|
||||
|
|
|
|||
|
|
@ -75,9 +75,9 @@ public:
|
|||
|
||||
bool isCached() const override { return true; }
|
||||
|
||||
std::unique_ptr<IExternalLoadable> clone() const override
|
||||
std::shared_ptr<const IExternalLoadable> clone() const override
|
||||
{
|
||||
return std::make_unique<ComplexKeyCacheDictionary>(name, dict_struct, source_ptr->clone(), dict_lifetime, size);
|
||||
return std::make_shared<ComplexKeyCacheDictionary>(name, dict_struct, source_ptr->clone(), dict_lifetime, size);
|
||||
}
|
||||
|
||||
const IDictionarySource * getSource() const override { return source_ptr.get(); }
|
||||
|
|
|
|||
|
|
@ -50,9 +50,9 @@ public:
|
|||
|
||||
bool isCached() const override { return false; }
|
||||
|
||||
std::unique_ptr<IExternalLoadable> clone() const override
|
||||
std::shared_ptr<const IExternalLoadable> clone() const override
|
||||
{
|
||||
return std::make_unique<ComplexKeyHashedDictionary>(name, dict_struct, source_ptr->clone(), dict_lifetime, require_nonempty, saved_block);
|
||||
return std::make_shared<ComplexKeyHashedDictionary>(name, dict_struct, source_ptr->clone(), dict_lifetime, require_nonempty, saved_block);
|
||||
}
|
||||
|
||||
const IDictionarySource * getSource() const override { return source_ptr.get(); }
|
||||
|
|
|
|||
|
|
@ -88,17 +88,14 @@ DictionarySourcePtr DictionarySourceFactory::create(
|
|||
throw Exception{name + ": element dictionary.source should have exactly one child element",
|
||||
ErrorCodes::EXCESSIVE_ELEMENT_IN_CONFIG};
|
||||
|
||||
auto sample_block = createSampleBlock(dict_struct);
|
||||
|
||||
const auto & source_type = keys.front();
|
||||
|
||||
const auto found = registered_sources.find(source_type);
|
||||
if (found != registered_sources.end())
|
||||
{
|
||||
const auto found = registered_sources.find(source_type);
|
||||
if (found != registered_sources.end())
|
||||
{
|
||||
const auto & create_source = found->second;
|
||||
return create_source(dict_struct, config, config_prefix, sample_block, context);
|
||||
}
|
||||
const auto & create_source = found->second;
|
||||
auto sample_block = createSampleBlock(dict_struct);
|
||||
return create_source(dict_struct, config, config_prefix, sample_block, context);
|
||||
}
|
||||
|
||||
throw Exception{name + ": unknown dictionary source type: " + source_type, ErrorCodes::UNKNOWN_ELEMENT_IN_CONFIG};
|
||||
|
|
|
|||
|
|
@ -47,9 +47,9 @@ public:
|
|||
|
||||
bool isCached() const override { return false; }
|
||||
|
||||
std::unique_ptr<IExternalLoadable> clone() const override
|
||||
std::shared_ptr<const IExternalLoadable> clone() const override
|
||||
{
|
||||
return std::make_unique<FlatDictionary>(name, dict_struct, source_ptr->clone(), dict_lifetime, require_nonempty, saved_block);
|
||||
return std::make_shared<FlatDictionary>(name, dict_struct, source_ptr->clone(), dict_lifetime, require_nonempty, saved_block);
|
||||
}
|
||||
|
||||
const IDictionarySource * getSource() const override { return source_ptr.get(); }
|
||||
|
|
|
|||
|
|
@ -46,9 +46,9 @@ public:
|
|||
|
||||
bool isCached() const override { return false; }
|
||||
|
||||
std::unique_ptr<IExternalLoadable> clone() const override
|
||||
std::shared_ptr<const IExternalLoadable> clone() const override
|
||||
{
|
||||
return std::make_unique<HashedDictionary>(name, dict_struct, source_ptr->clone(), dict_lifetime, require_nonempty, saved_block);
|
||||
return std::make_shared<HashedDictionary>(name, dict_struct, source_ptr->clone(), dict_lifetime, require_nonempty, saved_block);
|
||||
}
|
||||
|
||||
const IDictionarySource * getSource() const override { return source_ptr.get(); }
|
||||
|
|
|
|||
|
|
@ -42,9 +42,9 @@ public:
|
|||
|
||||
bool isCached() const override { return false; }
|
||||
|
||||
std::unique_ptr<IExternalLoadable> clone() const override
|
||||
std::shared_ptr<const IExternalLoadable> clone() const override
|
||||
{
|
||||
return std::make_unique<RangeHashedDictionary>(dictionary_name, dict_struct, source_ptr->clone(), dict_lifetime, require_nonempty);
|
||||
return std::make_shared<RangeHashedDictionary>(dictionary_name, dict_struct, source_ptr->clone(), dict_lifetime, require_nonempty);
|
||||
}
|
||||
|
||||
const IDictionarySource * getSource() const override { return source_ptr.get(); }
|
||||
|
|
|
|||
|
|
@ -51,9 +51,9 @@ public:
|
|||
|
||||
bool isCached() const override { return false; }
|
||||
|
||||
std::unique_ptr<IExternalLoadable> clone() const override
|
||||
std::shared_ptr<const IExternalLoadable> clone() const override
|
||||
{
|
||||
return std::make_unique<TrieDictionary>(name, dict_struct, source_ptr->clone(), dict_lifetime, require_nonempty);
|
||||
return std::make_shared<TrieDictionary>(name, dict_struct, source_ptr->clone(), dict_lifetime, require_nonempty);
|
||||
}
|
||||
|
||||
const IDictionarySource * getSource() const override { return source_ptr.get(); }
|
||||
|
|
|
|||
|
|
@ -537,9 +537,9 @@ bool CatBoostModel::isModified() const
|
|||
return true;
|
||||
}
|
||||
|
||||
std::unique_ptr<IExternalLoadable> CatBoostModel::clone() const
|
||||
std::shared_ptr<const IExternalLoadable> CatBoostModel::clone() const
|
||||
{
|
||||
return std::make_unique<CatBoostModel>(name, model_path, lib_path, lifetime);
|
||||
return std::make_shared<CatBoostModel>(name, model_path, lib_path, lifetime);
|
||||
}
|
||||
|
||||
size_t CatBoostModel::getFloatFeaturesCount() const
|
||||
|
|
|
|||
|
|
@ -66,7 +66,7 @@ public:
|
|||
|
||||
bool isModified() const override;
|
||||
|
||||
std::unique_ptr<IExternalLoadable> clone() const override;
|
||||
std::shared_ptr<const IExternalLoadable> clone() const override;
|
||||
|
||||
std::chrono::time_point<std::chrono::system_clock> getCreationTime() const override { return creation_time; }
|
||||
std::exception_ptr getCreationException() const override { return creation_exception; }
|
||||
|
|
|
|||
|
|
@ -105,7 +105,7 @@ struct ContextShared
|
|||
mutable std::recursive_mutex mutex;
|
||||
/// Separate mutex for access of dictionaries. Separate mutex to avoid locks when server doing request to itself.
|
||||
mutable std::mutex embedded_dictionaries_mutex;
|
||||
mutable std::recursive_mutex external_dictionaries_mutex;
|
||||
mutable std::mutex external_dictionaries_mutex;
|
||||
mutable std::mutex external_models_mutex;
|
||||
/// Separate mutex for re-initialization of zookeer session. This operation could take a long time and must not interfere with another operations.
|
||||
mutable std::mutex zookeeper_mutex;
|
||||
|
|
@ -892,27 +892,34 @@ StoragePtr Context::tryGetTable(const String & database_name, const String & tab
|
|||
|
||||
StoragePtr Context::getTableImpl(const String & database_name, const String & table_name, Exception * exception) const
|
||||
{
|
||||
auto lock = getLock();
|
||||
String db;
|
||||
DatabasePtr database;
|
||||
|
||||
if (database_name.empty())
|
||||
{
|
||||
StoragePtr res = tryGetExternalTable(table_name);
|
||||
if (res)
|
||||
return res;
|
||||
auto lock = getLock();
|
||||
|
||||
if (database_name.empty())
|
||||
{
|
||||
StoragePtr res = tryGetExternalTable(table_name);
|
||||
if (res)
|
||||
return res;
|
||||
}
|
||||
|
||||
db = resolveDatabase(database_name, current_database);
|
||||
checkDatabaseAccessRightsImpl(db);
|
||||
|
||||
Databases::const_iterator it = shared->databases.find(db);
|
||||
if (shared->databases.end() == it)
|
||||
{
|
||||
if (exception)
|
||||
*exception = Exception("Database " + backQuoteIfNeed(db) + " doesn't exist", ErrorCodes::UNKNOWN_DATABASE);
|
||||
return {};
|
||||
}
|
||||
|
||||
database = it->second;
|
||||
}
|
||||
|
||||
String db = resolveDatabase(database_name, current_database);
|
||||
checkDatabaseAccessRightsImpl(db);
|
||||
|
||||
Databases::const_iterator it = shared->databases.find(db);
|
||||
if (shared->databases.end() == it)
|
||||
{
|
||||
if (exception)
|
||||
*exception = Exception("Database " + backQuoteIfNeed(db) + " doesn't exist", ErrorCodes::UNKNOWN_DATABASE);
|
||||
return {};
|
||||
}
|
||||
|
||||
auto table = it->second->tryGetTable(*this, table_name);
|
||||
auto table = database->tryGetTable(*this, table_name);
|
||||
if (!table)
|
||||
{
|
||||
if (exception)
|
||||
|
|
@ -1253,23 +1260,48 @@ EmbeddedDictionaries & Context::getEmbeddedDictionaries()
|
|||
|
||||
const ExternalDictionaries & Context::getExternalDictionaries() const
|
||||
{
|
||||
return getExternalDictionariesImpl(false);
|
||||
{
|
||||
std::lock_guard lock(shared->external_dictionaries_mutex);
|
||||
if (shared->external_dictionaries)
|
||||
return *shared->external_dictionaries;
|
||||
}
|
||||
|
||||
const auto & config = getConfigRef();
|
||||
std::lock_guard lock(shared->external_dictionaries_mutex);
|
||||
if (!shared->external_dictionaries)
|
||||
{
|
||||
if (!this->global_context)
|
||||
throw Exception("Logical error: there is no global context", ErrorCodes::LOGICAL_ERROR);
|
||||
|
||||
auto config_repository = shared->runtime_components_factory->createExternalDictionariesConfigRepository();
|
||||
shared->external_dictionaries.emplace(std::move(config_repository), config, *this->global_context);
|
||||
}
|
||||
return *shared->external_dictionaries;
|
||||
}
|
||||
|
||||
ExternalDictionaries & Context::getExternalDictionaries()
|
||||
{
|
||||
return getExternalDictionariesImpl(false);
|
||||
return const_cast<ExternalDictionaries &>(const_cast<const Context *>(this)->getExternalDictionaries());
|
||||
}
|
||||
|
||||
|
||||
const ExternalModels & Context::getExternalModels() const
|
||||
{
|
||||
return getExternalModelsImpl(false);
|
||||
std::lock_guard lock(shared->external_models_mutex);
|
||||
if (!shared->external_models)
|
||||
{
|
||||
if (!this->global_context)
|
||||
throw Exception("Logical error: there is no global context", ErrorCodes::LOGICAL_ERROR);
|
||||
|
||||
auto config_repository = shared->runtime_components_factory->createExternalModelsConfigRepository();
|
||||
shared->external_models.emplace(std::move(config_repository), *this->global_context);
|
||||
}
|
||||
return *shared->external_models;
|
||||
}
|
||||
|
||||
ExternalModels & Context::getExternalModels()
|
||||
{
|
||||
return getExternalModelsImpl(false);
|
||||
return const_cast<ExternalModels &>(const_cast<const Context *>(this)->getExternalModels());
|
||||
}
|
||||
|
||||
|
||||
|
|
@ -1291,61 +1323,12 @@ EmbeddedDictionaries & Context::getEmbeddedDictionariesImpl(const bool throw_on_
|
|||
}
|
||||
|
||||
|
||||
ExternalDictionaries & Context::getExternalDictionariesImpl(const bool throw_on_error) const
|
||||
{
|
||||
{
|
||||
std::lock_guard lock(shared->external_dictionaries_mutex);
|
||||
if (shared->external_dictionaries)
|
||||
return *shared->external_dictionaries;
|
||||
}
|
||||
|
||||
const auto & config = getConfigRef();
|
||||
std::lock_guard lock(shared->external_dictionaries_mutex);
|
||||
if (!shared->external_dictionaries)
|
||||
{
|
||||
if (!this->global_context)
|
||||
throw Exception("Logical error: there is no global context", ErrorCodes::LOGICAL_ERROR);
|
||||
|
||||
auto config_repository = shared->runtime_components_factory->createExternalDictionariesConfigRepository();
|
||||
shared->external_dictionaries.emplace(std::move(config_repository), config, *this->global_context);
|
||||
shared->external_dictionaries->init(throw_on_error);
|
||||
}
|
||||
return *shared->external_dictionaries;
|
||||
}
|
||||
|
||||
ExternalModels & Context::getExternalModelsImpl(bool throw_on_error) const
|
||||
{
|
||||
std::lock_guard lock(shared->external_models_mutex);
|
||||
if (!shared->external_models)
|
||||
{
|
||||
if (!this->global_context)
|
||||
throw Exception("Logical error: there is no global context", ErrorCodes::LOGICAL_ERROR);
|
||||
|
||||
auto config_repository = shared->runtime_components_factory->createExternalModelsConfigRepository();
|
||||
shared->external_models.emplace(std::move(config_repository), *this->global_context);
|
||||
shared->external_models->init(throw_on_error);
|
||||
}
|
||||
return *shared->external_models;
|
||||
}
|
||||
|
||||
void Context::tryCreateEmbeddedDictionaries() const
|
||||
{
|
||||
static_cast<void>(getEmbeddedDictionariesImpl(true));
|
||||
}
|
||||
|
||||
|
||||
void Context::tryCreateExternalDictionaries() const
|
||||
{
|
||||
static_cast<void>(getExternalDictionariesImpl(true));
|
||||
}
|
||||
|
||||
|
||||
void Context::tryCreateExternalModels() const
|
||||
{
|
||||
static_cast<void>(getExternalModelsImpl(true));
|
||||
}
|
||||
|
||||
|
||||
void Context::setProgressCallback(ProgressCallback callback)
|
||||
{
|
||||
/// Callback is set to a session or to a query. In the session, only one query is processed at a time. Therefore, the lock is not needed.
|
||||
|
|
|
|||
|
|
@ -282,8 +282,6 @@ public:
|
|||
ExternalDictionaries & getExternalDictionaries();
|
||||
ExternalModels & getExternalModels();
|
||||
void tryCreateEmbeddedDictionaries() const;
|
||||
void tryCreateExternalDictionaries() const;
|
||||
void tryCreateExternalModels() const;
|
||||
|
||||
/// I/O formats.
|
||||
BlockInputStreamPtr getInputFormat(const String & name, ReadBuffer & buf, const Block & sample, UInt64 max_block_size) const;
|
||||
|
|
@ -493,8 +491,6 @@ private:
|
|||
void setProfile(const String & profile);
|
||||
|
||||
EmbeddedDictionaries & getEmbeddedDictionariesImpl(bool throw_on_error) const;
|
||||
ExternalDictionaries & getExternalDictionariesImpl(bool throw_on_error) const;
|
||||
ExternalModels & getExternalModelsImpl(bool throw_on_error) const;
|
||||
|
||||
StoragePtr getTableImpl(const String & database_name, const String & table_name, Exception * exception) const;
|
||||
|
||||
|
|
|
|||
|
|
@ -5,47 +5,26 @@
|
|||
namespace DB
|
||||
{
|
||||
|
||||
namespace
|
||||
{
|
||||
const ExternalLoaderUpdateSettings externalDictionariesUpdateSettings {};
|
||||
|
||||
const ExternalLoaderConfigSettings & getExternalDictionariesConfigSettings()
|
||||
{
|
||||
static ExternalLoaderConfigSettings settings;
|
||||
static std::once_flag flag;
|
||||
|
||||
std::call_once(flag, []
|
||||
{
|
||||
settings.external_config = "dictionary";
|
||||
settings.external_name = "name";
|
||||
settings.path_setting_name = "dictionaries_config";
|
||||
});
|
||||
|
||||
return settings;
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
/// Must not acquire Context lock in constructor to avoid possibility of deadlocks.
|
||||
ExternalDictionaries::ExternalDictionaries(
|
||||
std::unique_ptr<IExternalLoaderConfigRepository> config_repository,
|
||||
const Poco::Util::AbstractConfiguration & config,
|
||||
Context & context)
|
||||
: ExternalLoader(config,
|
||||
externalDictionariesUpdateSettings,
|
||||
getExternalDictionariesConfigSettings(),
|
||||
std::move(config_repository),
|
||||
&Logger::get("ExternalDictionaries"),
|
||||
"external dictionary"),
|
||||
"external dictionary",
|
||||
&Logger::get("ExternalDictionaries")),
|
||||
context(context)
|
||||
{
|
||||
addConfigRepository(std::move(config_repository), {"dictionary", "name", "dictionaries_config"});
|
||||
enableAsyncLoading(true);
|
||||
enablePeriodicUpdates(true);
|
||||
}
|
||||
|
||||
|
||||
std::unique_ptr<IExternalLoadable> ExternalDictionaries::create(
|
||||
const std::string & name, const Configuration & config, const std::string & config_prefix) const
|
||||
ExternalLoader::LoadablePtr ExternalDictionaries::create(
|
||||
const std::string & name, const Poco::Util::AbstractConfiguration & config, const std::string & key_in_config) const
|
||||
{
|
||||
return DictionaryFactory::instance().create(name, config, config_prefix, context);
|
||||
return DictionaryFactory::instance().create(name, config, key_in_config, context);
|
||||
}
|
||||
|
||||
}
|
||||
|
|
|
|||
|
|
@ -15,7 +15,7 @@ class Context;
|
|||
class ExternalDictionaries : public ExternalLoader
|
||||
{
|
||||
public:
|
||||
using DictPtr = std::shared_ptr<IDictionaryBase>;
|
||||
using DictPtr = std::shared_ptr<const IDictionaryBase>;
|
||||
|
||||
/// Dictionaries will be loaded immediately and then will be updated in separate thread, each 'reload_period' seconds.
|
||||
ExternalDictionaries(
|
||||
|
|
@ -23,25 +23,19 @@ public:
|
|||
const Poco::Util::AbstractConfiguration & config,
|
||||
Context & context);
|
||||
|
||||
/// Forcibly reloads specified dictionary.
|
||||
void reloadDictionary(const std::string & name) { reload(name); }
|
||||
|
||||
DictPtr getDictionary(const std::string & name) const
|
||||
{
|
||||
return std::static_pointer_cast<IDictionaryBase>(getLoadable(name));
|
||||
return std::static_pointer_cast<const IDictionaryBase>(getLoadable(name));
|
||||
}
|
||||
|
||||
DictPtr tryGetDictionary(const std::string & name) const
|
||||
{
|
||||
return std::static_pointer_cast<IDictionaryBase>(tryGetLoadable(name));
|
||||
return std::static_pointer_cast<const IDictionaryBase>(tryGetLoadable(name));
|
||||
}
|
||||
|
||||
protected:
|
||||
|
||||
std::unique_ptr<IExternalLoadable> create(const std::string & name, const Configuration & config,
|
||||
const std::string & config_prefix) const override;
|
||||
|
||||
using ExternalLoader::getObjectsMap;
|
||||
LoadablePtr create(const std::string & name, const Poco::Util::AbstractConfiguration & config,
|
||||
const std::string & key_in_config) const override;
|
||||
|
||||
friend class StorageSystemDictionaries;
|
||||
friend class DatabaseDictionary;
|
||||
|
|
|
|||
File diff suppressed because it is too large
Load Diff
|
|
@ -1,24 +1,16 @@
|
|||
#pragma once
|
||||
|
||||
#include <common/logger_useful.h>
|
||||
#include <Poco/Event.h>
|
||||
#include <mutex>
|
||||
#include <thread>
|
||||
#include <unordered_map>
|
||||
#include <unordered_set>
|
||||
#include <chrono>
|
||||
#include <tuple>
|
||||
#include <functional>
|
||||
#include <unordered_map>
|
||||
#include <Core/Types.h>
|
||||
#include <Interpreters/IExternalLoadable.h>
|
||||
#include <Interpreters/IExternalLoaderConfigRepository.h>
|
||||
#include <Core/Types.h>
|
||||
#include <pcg_random.hpp>
|
||||
#include <Common/randomSeed.h>
|
||||
#include <Common/ThreadPool.h>
|
||||
#include <common/logger_useful.h>
|
||||
|
||||
|
||||
namespace DB
|
||||
{
|
||||
|
||||
struct ExternalLoaderUpdateSettings
|
||||
{
|
||||
UInt64 check_period_sec = 5;
|
||||
|
|
@ -28,9 +20,7 @@ struct ExternalLoaderUpdateSettings
|
|||
|
||||
ExternalLoaderUpdateSettings() = default;
|
||||
ExternalLoaderUpdateSettings(UInt64 check_period_sec, UInt64 backoff_initial_sec, UInt64 backoff_max_sec)
|
||||
: check_period_sec(check_period_sec),
|
||||
backoff_initial_sec(backoff_initial_sec),
|
||||
backoff_max_sec(backoff_max_sec) {}
|
||||
: check_period_sec(check_period_sec), backoff_initial_sec(backoff_initial_sec), backoff_max_sec(backoff_max_sec) {}
|
||||
};
|
||||
|
||||
|
||||
|
|
@ -51,154 +41,171 @@ struct ExternalLoaderConfigSettings
|
|||
std::string path_setting_name;
|
||||
};
|
||||
|
||||
|
||||
/** Manages user-defined objects.
|
||||
* Monitors configuration file and automatically reloads objects in a separate thread.
|
||||
* The monitoring thread wakes up every 'check_period_sec' seconds and checks
|
||||
* modification time of objects' configuration file. If said time is greater than
|
||||
* 'config_last_modified', the objects are created from scratch using configuration file,
|
||||
* possibly overriding currently existing objects with the same name (previous versions of
|
||||
* overridden objects will live as long as there are any users retaining them).
|
||||
*
|
||||
* Apart from checking configuration file for modifications, each object
|
||||
* has a lifetime of its own and may be updated if it supportUpdates.
|
||||
* The time of next update is calculated by choosing uniformly a random number
|
||||
* distributed between lifetime.min_sec and lifetime.max_sec.
|
||||
* If either of lifetime.min_sec and lifetime.max_sec is zero, such object is never updated.
|
||||
*/
|
||||
* Monitors configuration file and automatically reloads objects in separate threads.
|
||||
* The monitoring thread wakes up every 'check_period_sec' seconds and checks
|
||||
* modification time of objects' configuration file. If said time is greater than
|
||||
* 'config_last_modified', the objects are created from scratch using configuration file,
|
||||
* possibly overriding currently existing objects with the same name (previous versions of
|
||||
* overridden objects will live as long as there are any users retaining them).
|
||||
*
|
||||
* Apart from checking configuration file for modifications, each object
|
||||
* has a lifetime of its own and may be updated if it supportUpdates.
|
||||
* The time of next update is calculated by choosing uniformly a random number
|
||||
* distributed between lifetime.min_sec and lifetime.max_sec.
|
||||
* If either of lifetime.min_sec and lifetime.max_sec is zero, such object is never updated.
|
||||
*/
|
||||
class ExternalLoader
|
||||
{
|
||||
public:
|
||||
using LoadablePtr = std::shared_ptr<IExternalLoadable>;
|
||||
using LoadablePtr = std::shared_ptr<const IExternalLoadable>;
|
||||
using Loadables = std::vector<LoadablePtr>;
|
||||
|
||||
private:
|
||||
struct LoadableInfo final
|
||||
enum class Status
|
||||
{
|
||||
LoadablePtr loadable;
|
||||
std::string origin;
|
||||
NOT_LOADED, /// Object hasn't been tried to load. This is an initial state.
|
||||
LOADED, /// Object has been loaded successfully.
|
||||
FAILED, /// Object has been failed to load.
|
||||
LOADING, /// Object is being loaded right now for the first time.
|
||||
FAILED_AND_RELOADING, /// Object was failed to load before and it's being reloaded right now.
|
||||
LOADED_AND_RELOADING, /// Object was loaded successfully before and it's being reloaded right now.
|
||||
NOT_EXIST, /// Object with this name wasn't found in the configuration.
|
||||
};
|
||||
|
||||
static std::vector<std::pair<String, Int8>> getStatusEnumAllPossibleValues();
|
||||
|
||||
using Duration = std::chrono::milliseconds;
|
||||
using TimePoint = std::chrono::system_clock::time_point;
|
||||
|
||||
struct LoadResult
|
||||
{
|
||||
LoadResult(Status status_) : status(status_) {}
|
||||
Status status;
|
||||
LoadablePtr object;
|
||||
String origin;
|
||||
TimePoint loading_start_time;
|
||||
Duration loading_duration;
|
||||
std::exception_ptr exception;
|
||||
};
|
||||
|
||||
struct FailedLoadableInfo final
|
||||
{
|
||||
std::unique_ptr<IExternalLoadable> loadable;
|
||||
std::chrono::system_clock::time_point next_attempt_time;
|
||||
UInt64 error_count;
|
||||
};
|
||||
using LoadResults = std::vector<std::pair<String, LoadResult>>;
|
||||
|
||||
public:
|
||||
using Configuration = Poco::Util::AbstractConfiguration;
|
||||
using ObjectsMap = std::unordered_map<std::string, LoadableInfo>;
|
||||
|
||||
/// Call init() after constructing the instance of any derived class.
|
||||
ExternalLoader(const Configuration & config_main,
|
||||
const ExternalLoaderUpdateSettings & update_settings,
|
||||
const ExternalLoaderConfigSettings & config_settings,
|
||||
std::unique_ptr<IExternalLoaderConfigRepository> config_repository,
|
||||
Logger * log, const std::string & loadable_object_name);
|
||||
ExternalLoader(const Poco::Util::AbstractConfiguration & main_config, const String & type_name_, Logger * log);
|
||||
virtual ~ExternalLoader();
|
||||
|
||||
/// Should be called after creating an instance of a derived class.
|
||||
/// Loads the objects immediately and starts a separate thread to update them once in each 'reload_period' seconds.
|
||||
/// This function does nothing if called again.
|
||||
void init(bool throw_on_error);
|
||||
/// Adds a repository which will be used to read configurations from.
|
||||
void addConfigRepository(
|
||||
std::unique_ptr<IExternalLoaderConfigRepository> config_repository, const ExternalLoaderConfigSettings & config_settings);
|
||||
|
||||
/// Forcibly reloads all loadable objects.
|
||||
void reload();
|
||||
/// Sets whether all the objects from the configuration should be always loaded (even those which are never used).
|
||||
void enableAlwaysLoadEverything(bool enable);
|
||||
|
||||
/// Forcibly reloads specified loadable object.
|
||||
void reload(const std::string & name);
|
||||
/// Sets whether the objects should be loaded asynchronously, each loading in a new thread (from the thread pool).
|
||||
void enableAsyncLoading(bool enable);
|
||||
|
||||
LoadablePtr getLoadable(const std::string & name) const;
|
||||
LoadablePtr tryGetLoadable(const std::string & name) const;
|
||||
/// Sets settings for periodic updates.
|
||||
void enablePeriodicUpdates(bool enable, const ExternalLoaderUpdateSettings & settings = {});
|
||||
|
||||
/// Returns the names of all the objects in the configuration (loaded or not).
|
||||
std::vector<String> getNames() const;
|
||||
size_t getNumberOfNames() const;
|
||||
|
||||
/// Returns the status of the object.
|
||||
/// If the object has not been loaded yet then the function returns Status::NOT_LOADED.
|
||||
/// If the specified name isn't found in the configuration then the function returns Status::NOT_EXIST.
|
||||
Status getCurrentStatus(const String & name) const;
|
||||
|
||||
/// Returns the result of loading the object.
|
||||
/// The function doesn't load anything, it just returns the current load result as is.
|
||||
LoadResult getCurrentLoadResult(const String & name) const;
|
||||
|
||||
using FilterByNameFunction = std::function<bool(const String &)>;
|
||||
|
||||
/// Returns all the load results as a map.
|
||||
/// The function doesn't load anything, it just returns the current load results as is.
|
||||
LoadResults getCurrentLoadResults() const;
|
||||
LoadResults getCurrentLoadResults(const FilterByNameFunction & filter_by_name) const;
|
||||
|
||||
/// Returns all loaded objects as a map.
|
||||
/// The function doesn't load anything, it just returns the current load results as is.
|
||||
Loadables getCurrentlyLoadedObjects() const;
|
||||
Loadables getCurrentlyLoadedObjects(const FilterByNameFunction & filter_by_name) const;
|
||||
size_t getNumberOfCurrentlyLoadedObjects() const;
|
||||
|
||||
static constexpr Duration NO_TIMEOUT = Duration::max();
|
||||
|
||||
/// Starts loading of a specified object.
|
||||
void load(const String & name) const;
|
||||
|
||||
/// Tries to finish loading of a specified object during the timeout.
|
||||
/// Returns nullptr if the loading is unsuccessful or if there is no such object.
|
||||
void load(const String & name, LoadablePtr & loaded_object, Duration timeout = NO_TIMEOUT) const;
|
||||
void load(const String & name, LoadResult & load_result, Duration timeout = NO_TIMEOUT) const;
|
||||
LoadablePtr loadAndGet(const String & name, Duration timeout = NO_TIMEOUT) const { LoadablePtr object; load(name, object, timeout); return object; }
|
||||
LoadablePtr tryGetLoadable(const String & name) const { return loadAndGet(name); }
|
||||
|
||||
/// Tries to finish loading of a specified object during the timeout.
|
||||
/// Throws an exception if the loading is unsuccessful or if there is no such object.
|
||||
void loadStrict(const String & name, LoadablePtr & loaded_object) const;
|
||||
void loadStrict(const String & name, LoadResult & load_result) const;
|
||||
LoadablePtr loadAndGetStrict(const String & name) const { LoadablePtr object; loadStrict(name, object); return object; }
|
||||
LoadablePtr getLoadable(const String & name) const { return loadAndGetStrict(name); }
|
||||
|
||||
/// Tries to start loading of the objects for which the specified function returns true.
|
||||
void load(const FilterByNameFunction & filter_by_name) const;
|
||||
|
||||
/// Tries to finish loading of the objects for which the specified function returns true.
|
||||
void load(const FilterByNameFunction & filter_by_name, Loadables & loaded_objects, Duration timeout = NO_TIMEOUT) const;
|
||||
void load(const FilterByNameFunction & filter_by_name, LoadResults & load_results, Duration timeout = NO_TIMEOUT) const;
|
||||
Loadables loadAndGet(const FilterByNameFunction & filter_by_name, Duration timeout = NO_TIMEOUT) const { Loadables loaded_objects; load(filter_by_name, loaded_objects, timeout); return loaded_objects; }
|
||||
|
||||
/// Starts loading of all the objects.
|
||||
void load() const;
|
||||
|
||||
/// Tries to finish loading of all the objects during the timeout.
|
||||
void load(Loadables & loaded_objects, Duration timeout = NO_TIMEOUT) const;
|
||||
void load(LoadResults & load_results, Duration timeout = NO_TIMEOUT) const;
|
||||
|
||||
/// Starts reloading of a specified object.
|
||||
/// `load_never_loading` specifies what to do if the object has never been loading before.
|
||||
/// The function can either skip it (false) or load for the first time (true).
|
||||
void reload(const String & name, bool load_never_loading = false);
|
||||
|
||||
/// Starts reloading of the objects for which the specified function returns true.
|
||||
/// `load_never_loading` specifies what to do with the objects which have never been loading before.
|
||||
/// The function can either skip them (false) or load for the first time (true).
|
||||
void reload(const FilterByNameFunction & filter_by_name, bool load_never_loading = false);
|
||||
|
||||
/// Starts reloading of all the objects.
|
||||
/// `load_never_loading` specifies what to do with the objects which have never been loading before.
|
||||
/// The function can either skip them (false) or load for the first time (true).
|
||||
void reload(bool load_never_loading = false);
|
||||
|
||||
protected:
|
||||
virtual std::unique_ptr<IExternalLoadable> create(const std::string & name, const Configuration & config,
|
||||
const std::string & config_prefix) const = 0;
|
||||
|
||||
class LockedObjectsMap
|
||||
{
|
||||
public:
|
||||
LockedObjectsMap(std::mutex & mutex, const ObjectsMap & objects_map) : lock(mutex), objects_map(objects_map) {}
|
||||
const ObjectsMap & get() { return objects_map; }
|
||||
private:
|
||||
std::unique_lock<std::mutex> lock;
|
||||
const ObjectsMap & objects_map;
|
||||
};
|
||||
|
||||
/// Direct access to objects.
|
||||
LockedObjectsMap getObjectsMap() const;
|
||||
virtual LoadablePtr create(const String & name, const Poco::Util::AbstractConfiguration & config, const String & key_in_config) const = 0;
|
||||
|
||||
private:
|
||||
std::once_flag is_initialized_flag;
|
||||
struct ObjectConfig;
|
||||
using ObjectWithException = std::pair<LoadablePtr, std::exception_ptr>;
|
||||
|
||||
/// Protects only objects map.
|
||||
/** Reading and assignment of "loadable" should be done under mutex.
|
||||
* Creating new versions of "loadable" should not be done under mutex.
|
||||
*/
|
||||
mutable std::mutex map_mutex;
|
||||
ObjectWithException
|
||||
createObject(const String & name, const ObjectConfig & config, bool config_changed, const LoadablePtr & previous_version) const;
|
||||
TimePoint calculateNextUpdateTime(const LoadablePtr & loaded_object, size_t error_count) const;
|
||||
|
||||
/// Protects all data, currently used to avoid races between updating thread and SYSTEM queries.
|
||||
/// The mutex is recursive because creating of objects might be recursive, i.e.
|
||||
/// creating objects might cause creating other objects.
|
||||
mutable std::recursive_mutex all_mutex;
|
||||
class ConfigFilesReader;
|
||||
std::unique_ptr<ConfigFilesReader> config_files_reader;
|
||||
|
||||
/// name -> loadable.
|
||||
mutable ObjectsMap loadable_objects;
|
||||
class LoadingDispatcher;
|
||||
std::unique_ptr<LoadingDispatcher> loading_dispatcher;
|
||||
|
||||
struct LoadableCreationInfo
|
||||
{
|
||||
std::string name;
|
||||
Poco::AutoPtr<Poco::Util::AbstractConfiguration> config;
|
||||
std::string config_path;
|
||||
std::string config_prefix;
|
||||
};
|
||||
class PeriodicUpdater;
|
||||
std::unique_ptr<PeriodicUpdater> periodic_updater;
|
||||
|
||||
/// Objects which should be reloaded soon.
|
||||
mutable std::unordered_map<std::string, LoadableCreationInfo> objects_to_reload;
|
||||
|
||||
/// Here are loadable objects, that has been never loaded successfully.
|
||||
/// They are also in 'loadable_objects', but with nullptr as 'loadable'.
|
||||
mutable std::unordered_map<std::string, FailedLoadableInfo> failed_loadable_objects;
|
||||
|
||||
/// Both for loadable_objects and failed_loadable_objects.
|
||||
mutable std::unordered_map<std::string, std::chrono::system_clock::time_point> update_times;
|
||||
|
||||
std::unordered_map<std::string, std::unordered_set<std::string>> loadable_objects_defined_in_config;
|
||||
|
||||
mutable pcg64 rnd_engine{randomSeed()};
|
||||
|
||||
const Configuration & config_main;
|
||||
const ExternalLoaderUpdateSettings & update_settings;
|
||||
const ExternalLoaderConfigSettings & config_settings;
|
||||
|
||||
std::unique_ptr<IExternalLoaderConfigRepository> config_repository;
|
||||
|
||||
ThreadFromGlobalPool reloading_thread;
|
||||
Poco::Event destroy;
|
||||
|
||||
Logger * log;
|
||||
/// Loadable object name to use in log messages.
|
||||
std::string object_name;
|
||||
|
||||
std::unordered_map<std::string, Poco::Timestamp> last_modification_times;
|
||||
|
||||
void initImpl(bool throw_on_error);
|
||||
|
||||
/// Check objects definitions in config files and reload or/and add new ones if the definition is changed
|
||||
/// If loadable_name is not empty, load only loadable object with name loadable_name
|
||||
void reloadFromConfigFiles(bool throw_on_error, bool force_reload = false, const std::string & loadable_name = "");
|
||||
void reloadFromConfigFile(const std::string & config_path, const bool force_reload, const std::string & loadable_name);
|
||||
|
||||
/// Check config files and update expired loadable objects
|
||||
void reloadAndUpdate(bool throw_on_error = false);
|
||||
|
||||
void reloadPeriodically();
|
||||
|
||||
void finishReload(const std::string & loadable_name, bool throw_on_error) const;
|
||||
void finishAllReloads(bool throw_on_error) const;
|
||||
void finishReloadImpl(const LoadableCreationInfo & creation_info, bool throw_on_error) const;
|
||||
|
||||
LoadablePtr getLoadableImpl(const std::string & name, bool throw_on_error) const;
|
||||
const String type_name;
|
||||
};
|
||||
|
||||
String toString(ExternalLoader::Status status);
|
||||
std::ostream & operator<<(std::ostream & out, ExternalLoader::Status status);
|
||||
|
||||
}
|
||||
|
|
|
|||
|
|
@ -9,43 +9,21 @@ namespace ErrorCodes
|
|||
extern const int INVALID_CONFIG_PARAMETER;
|
||||
}
|
||||
|
||||
namespace
|
||||
{
|
||||
|
||||
const ExternalLoaderUpdateSettings externalModelsUpdateSettings { };
|
||||
|
||||
const ExternalLoaderConfigSettings & getExternalModelsConfigSettings()
|
||||
{
|
||||
static ExternalLoaderConfigSettings settings;
|
||||
static std::once_flag flag;
|
||||
|
||||
std::call_once(flag, []
|
||||
{
|
||||
settings.external_config = "model";
|
||||
settings.external_name = "name";
|
||||
settings.path_setting_name = "models_config";
|
||||
});
|
||||
|
||||
return settings;
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
ExternalModels::ExternalModels(
|
||||
std::unique_ptr<IExternalLoaderConfigRepository> config_repository,
|
||||
Context & context)
|
||||
: ExternalLoader(context.getConfigRef(),
|
||||
externalModelsUpdateSettings,
|
||||
getExternalModelsConfigSettings(),
|
||||
std::move(config_repository),
|
||||
&Logger::get("ExternalModels"),
|
||||
"external model"),
|
||||
"external model",
|
||||
&Logger::get("ExternalModels")),
|
||||
context(context)
|
||||
{
|
||||
addConfigRepository(std::move(config_repository), {"model", "name", "models_config"});
|
||||
enablePeriodicUpdates(true);
|
||||
}
|
||||
|
||||
std::unique_ptr<IExternalLoadable> ExternalModels::create(
|
||||
const std::string & name, const Configuration & config, const std::string & config_prefix) const
|
||||
std::shared_ptr<const IExternalLoadable> ExternalModels::create(
|
||||
const std::string & name, const Poco::Util::AbstractConfiguration & config, const std::string & config_prefix) const
|
||||
{
|
||||
String type = config.getString(config_prefix + ".type");
|
||||
ExternalLoadableLifetime lifetime(config, config_prefix + ".lifetime");
|
||||
|
|
|
|||
|
|
@ -15,27 +15,21 @@ class Context;
|
|||
class ExternalModels : public ExternalLoader
|
||||
{
|
||||
public:
|
||||
using ModelPtr = std::shared_ptr<IModel>;
|
||||
using ModelPtr = std::shared_ptr<const IModel>;
|
||||
|
||||
/// Models will be loaded immediately and then will be updated in separate thread, each 'reload_period' seconds.
|
||||
ExternalModels(
|
||||
std::unique_ptr<IExternalLoaderConfigRepository> config_repository,
|
||||
Context & context);
|
||||
|
||||
/// Forcibly reloads specified model.
|
||||
void reloadModel(const std::string & name) { reload(name); }
|
||||
|
||||
ModelPtr getModel(const std::string & name) const
|
||||
{
|
||||
return std::static_pointer_cast<IModel>(getLoadable(name));
|
||||
return std::static_pointer_cast<const IModel>(getLoadable(name));
|
||||
}
|
||||
|
||||
protected:
|
||||
|
||||
std::unique_ptr<IExternalLoadable> create(const std::string & name, const Configuration & config,
|
||||
const std::string & config_prefix) const override;
|
||||
|
||||
using ExternalLoader::getObjectsMap;
|
||||
LoadablePtr create(const std::string & name, const Poco::Util::AbstractConfiguration & config,
|
||||
const std::string & key_in_config) const override;
|
||||
|
||||
friend class StorageSystemModels;
|
||||
private:
|
||||
|
|
|
|||
|
|
@ -0,0 +1,19 @@
|
|||
#include <Interpreters/IExternalLoadable.h>
|
||||
|
||||
#include <Poco/Util/AbstractConfiguration.h>
|
||||
|
||||
|
||||
namespace DB
|
||||
{
|
||||
|
||||
ExternalLoadableLifetime::ExternalLoadableLifetime(const Poco::Util::AbstractConfiguration & config,
|
||||
const std::string & config_prefix)
|
||||
{
|
||||
const auto & lifetime_min_key = config_prefix + ".min";
|
||||
const auto has_min = config.has(lifetime_min_key);
|
||||
|
||||
min_sec = has_min ? config.getUInt64(lifetime_min_key) : config.getUInt64(config_prefix);
|
||||
max_sec = has_min ? config.getUInt64(config_prefix + ".max") : min_sec;
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -17,7 +17,7 @@ namespace DB
|
|||
{
|
||||
|
||||
/// Min and max lifetimes for a loadable object or it's entry
|
||||
struct ExternalLoadableLifetime final
|
||||
struct ExternalLoadableLifetime
|
||||
{
|
||||
UInt64 min_sec;
|
||||
UInt64 max_sec;
|
||||
|
|
@ -40,7 +40,7 @@ public:
|
|||
/// If lifetime exceeded and isModified(), ExternalLoader replace current object with the result of clone().
|
||||
virtual bool isModified() const = 0;
|
||||
/// Returns new object with the same configuration. Is used to update modified object when lifetime exceeded.
|
||||
virtual std::unique_ptr<IExternalLoadable> clone() const = 0;
|
||||
virtual std::shared_ptr<const IExternalLoadable> clone() const = 0;
|
||||
|
||||
virtual std::chrono::time_point<std::chrono::system_clock> getCreationTime() const = 0;
|
||||
|
||||
|
|
|
|||
|
|
@ -156,7 +156,7 @@ BlockIO InterpreterSystemQuery::execute()
|
|||
break;
|
||||
#endif
|
||||
case Type::RELOAD_DICTIONARY:
|
||||
system_context.getExternalDictionaries().reloadDictionary(query.target_dictionary);
|
||||
system_context.getExternalDictionaries().reload(query.target_dictionary, true /* load the dictionary even if it wasn't loading before */);
|
||||
break;
|
||||
case Type::RELOAD_DICTIONARIES:
|
||||
executeCommandsAndThrowIfError(
|
||||
|
|
|
|||
|
|
@ -105,17 +105,13 @@ bool StorageMerge::hasColumn(const String & column_name) const
|
|||
template <typename F>
|
||||
StoragePtr StorageMerge::getFirstTable(F && predicate) const
|
||||
{
|
||||
auto database = global_context.getDatabase(source_database);
|
||||
auto iterator = database->getIterator(global_context);
|
||||
auto iterator = getDatabaseIterator(global_context);
|
||||
|
||||
while (iterator->isValid())
|
||||
{
|
||||
if (table_name_regexp.match(iterator->name()))
|
||||
{
|
||||
auto & table = iterator->table();
|
||||
if (table.get() != this && predicate(table))
|
||||
return table;
|
||||
}
|
||||
auto & table = iterator->table();
|
||||
if (table.get() != this && predicate(table))
|
||||
return table;
|
||||
|
||||
iterator->next();
|
||||
}
|
||||
|
|
@ -156,22 +152,17 @@ QueryProcessingStage::Enum StorageMerge::getQueryProcessingStage(const Context &
|
|||
{
|
||||
auto stage_in_source_tables = QueryProcessingStage::FetchColumns;
|
||||
|
||||
DatabasePtr database = context.getDatabase(source_database);
|
||||
DatabaseIteratorPtr iterator = database->getIterator(context);
|
||||
DatabaseIteratorPtr iterator = getDatabaseIterator(context);
|
||||
|
||||
size_t selected_table_size = 0;
|
||||
|
||||
while (iterator->isValid())
|
||||
{
|
||||
if (table_name_regexp.match(iterator->name()))
|
||||
auto & table = iterator->table();
|
||||
if (table.get() != this)
|
||||
{
|
||||
auto & table = iterator->table();
|
||||
if (table.get() != this)
|
||||
{
|
||||
++selected_table_size;
|
||||
stage_in_source_tables = std::max(stage_in_source_tables, table->getQueryProcessingStage(context));
|
||||
}
|
||||
|
||||
++selected_table_size;
|
||||
stage_in_source_tables = std::max(stage_in_source_tables, table->getQueryProcessingStage(context));
|
||||
}
|
||||
|
||||
iterator->next();
|
||||
|
|
@ -347,20 +338,17 @@ BlockInputStreams StorageMerge::createSourceStreams(const SelectQueryInfo & quer
|
|||
return source_streams;
|
||||
}
|
||||
|
||||
|
||||
StorageMerge::StorageListWithLocks StorageMerge::getSelectedTables(const String & query_id) const
|
||||
{
|
||||
StorageListWithLocks selected_tables;
|
||||
auto database = global_context.getDatabase(source_database);
|
||||
auto iterator = database->getIterator(global_context);
|
||||
auto iterator = getDatabaseIterator(global_context);
|
||||
|
||||
while (iterator->isValid())
|
||||
{
|
||||
if (table_name_regexp.match(iterator->name()))
|
||||
{
|
||||
auto & table = iterator->table();
|
||||
if (table.get() != this)
|
||||
selected_tables.emplace_back(table, table->lockStructureForShare(false, query_id));
|
||||
}
|
||||
auto & table = iterator->table();
|
||||
if (table.get() != this)
|
||||
selected_tables.emplace_back(table, table->lockStructureForShare(false, query_id));
|
||||
|
||||
iterator->next();
|
||||
}
|
||||
|
|
@ -372,25 +360,21 @@ StorageMerge::StorageListWithLocks StorageMerge::getSelectedTables(const String
|
|||
StorageMerge::StorageListWithLocks StorageMerge::getSelectedTables(const ASTPtr & query, bool has_virtual_column, bool get_lock, const String & query_id) const
|
||||
{
|
||||
StorageListWithLocks selected_tables;
|
||||
DatabasePtr database = global_context.getDatabase(source_database);
|
||||
DatabaseIteratorPtr iterator = database->getIterator(global_context);
|
||||
DatabaseIteratorPtr iterator = getDatabaseIterator(global_context);
|
||||
|
||||
auto virtual_column = ColumnString::create();
|
||||
|
||||
while (iterator->isValid())
|
||||
{
|
||||
if (table_name_regexp.match(iterator->name()))
|
||||
StoragePtr storage = iterator->table();
|
||||
|
||||
if (query && query->as<ASTSelectQuery>()->prewhere() && !storage->supportsPrewhere())
|
||||
throw Exception("Storage " + storage->getName() + " doesn't support PREWHERE.", ErrorCodes::ILLEGAL_PREWHERE);
|
||||
|
||||
if (storage.get() != this)
|
||||
{
|
||||
StoragePtr storage = iterator->table();
|
||||
|
||||
if (query && query->as<ASTSelectQuery>()->prewhere() && !storage->supportsPrewhere())
|
||||
throw Exception("Storage " + storage->getName() + " doesn't support PREWHERE.", ErrorCodes::ILLEGAL_PREWHERE);
|
||||
|
||||
if (storage.get() != this)
|
||||
{
|
||||
virtual_column->insert(storage->getTableName());
|
||||
selected_tables.emplace_back(storage, get_lock ? storage->lockStructureForShare(false, query_id) : TableStructureReadLockHolder{});
|
||||
}
|
||||
virtual_column->insert(storage->getTableName());
|
||||
selected_tables.emplace_back(storage, get_lock ? storage->lockStructureForShare(false, query_id) : TableStructureReadLockHolder{});
|
||||
}
|
||||
|
||||
iterator->next();
|
||||
|
|
@ -409,6 +393,15 @@ StorageMerge::StorageListWithLocks StorageMerge::getSelectedTables(const ASTPtr
|
|||
return selected_tables;
|
||||
}
|
||||
|
||||
|
||||
DatabaseIteratorPtr StorageMerge::getDatabaseIterator(const Context & context) const
|
||||
{
|
||||
auto database = context.getDatabase(source_database);
|
||||
auto table_name_match = [this](const String & table_name) { return table_name_regexp.match(table_name); };
|
||||
return database->getIterator(global_context, table_name_match);
|
||||
}
|
||||
|
||||
|
||||
void StorageMerge::alter(
|
||||
const AlterCommands & params, const String & database_name, const String & table_name,
|
||||
const Context & context, TableStructureWriteLockHolder & table_lock_holder)
|
||||
|
|
|
|||
|
|
@ -65,6 +65,8 @@ private:
|
|||
template <typename F>
|
||||
StoragePtr getFirstTable(F && predicate) const;
|
||||
|
||||
DatabaseIteratorPtr getDatabaseIterator(const Context & context) const;
|
||||
|
||||
protected:
|
||||
StorageMerge(
|
||||
const std::string & name_,
|
||||
|
|
|
|||
|
|
@ -2,6 +2,7 @@
|
|||
#include <DataTypes/DataTypeDateTime.h>
|
||||
#include <DataTypes/DataTypesNumber.h>
|
||||
#include <DataTypes/DataTypeString.h>
|
||||
#include <DataTypes/DataTypeEnum.h>
|
||||
#include <Dictionaries/IDictionary.h>
|
||||
#include <Dictionaries/IDictionarySource.h>
|
||||
#include <Dictionaries/DictionaryStructure.h>
|
||||
|
|
@ -19,6 +20,7 @@ NamesAndTypesList StorageSystemDictionaries::getNamesAndTypes()
|
|||
{
|
||||
return {
|
||||
{ "name", std::make_shared<DataTypeString>() },
|
||||
{ "status", std::make_shared<DataTypeEnum8>(ExternalLoader::getStatusEnumAllPossibleValues()) },
|
||||
{ "origin", std::make_shared<DataTypeString>() },
|
||||
{ "type", std::make_shared<DataTypeString>() },
|
||||
{ "key", std::make_shared<DataTypeString>() },
|
||||
|
|
@ -29,8 +31,10 @@ NamesAndTypesList StorageSystemDictionaries::getNamesAndTypes()
|
|||
{ "hit_rate", std::make_shared<DataTypeFloat64>() },
|
||||
{ "element_count", std::make_shared<DataTypeUInt64>() },
|
||||
{ "load_factor", std::make_shared<DataTypeFloat64>() },
|
||||
{ "creation_time", std::make_shared<DataTypeDateTime>() },
|
||||
{ "source", std::make_shared<DataTypeString>() },
|
||||
{ "loading_start_time", std::make_shared<DataTypeDateTime>() },
|
||||
{ "loading_duration", std::make_shared<DataTypeFloat32>() },
|
||||
//{ "creation_time", std::make_shared<DataTypeDateTime>() },
|
||||
{ "last_exception", std::make_shared<DataTypeString>() },
|
||||
};
|
||||
}
|
||||
|
|
@ -38,18 +42,17 @@ NamesAndTypesList StorageSystemDictionaries::getNamesAndTypes()
|
|||
void StorageSystemDictionaries::fillData(MutableColumns & res_columns, const Context & context, const SelectQueryInfo &) const
|
||||
{
|
||||
const auto & external_dictionaries = context.getExternalDictionaries();
|
||||
auto objects_map = external_dictionaries.getObjectsMap();
|
||||
const auto & dictionaries = objects_map.get();
|
||||
for (const auto & dict_info : dictionaries)
|
||||
for (const auto & [dict_name, load_result] : external_dictionaries.getCurrentLoadResults())
|
||||
{
|
||||
size_t i = 0;
|
||||
|
||||
res_columns[i++]->insert(dict_info.first);
|
||||
res_columns[i++]->insert(dict_info.second.origin);
|
||||
res_columns[i++]->insert(dict_name);
|
||||
res_columns[i++]->insert(static_cast<Int8>(load_result.status));
|
||||
res_columns[i++]->insert(load_result.origin);
|
||||
|
||||
if (dict_info.second.loadable)
|
||||
if (load_result.object)
|
||||
{
|
||||
const auto dict_ptr = std::static_pointer_cast<IDictionaryBase>(dict_info.second.loadable);
|
||||
const auto dict_ptr = std::static_pointer_cast<const IDictionaryBase>(load_result.object);
|
||||
|
||||
res_columns[i++]->insert(dict_ptr->getTypeName());
|
||||
|
||||
|
|
@ -62,26 +65,19 @@ void StorageSystemDictionaries::fillData(MutableColumns & res_columns, const Con
|
|||
res_columns[i++]->insert(dict_ptr->getHitRate());
|
||||
res_columns[i++]->insert(dict_ptr->getElementCount());
|
||||
res_columns[i++]->insert(dict_ptr->getLoadFactor());
|
||||
res_columns[i++]->insert(static_cast<UInt64>(std::chrono::system_clock::to_time_t(dict_ptr->getCreationTime())));
|
||||
res_columns[i++]->insert(dict_ptr->getSource()->toString());
|
||||
}
|
||||
else
|
||||
{
|
||||
while (i < 13)
|
||||
for (size_t j = 0; j != 10; ++j)
|
||||
res_columns[i++]->insertDefault();
|
||||
}
|
||||
|
||||
if (dict_info.second.exception)
|
||||
{
|
||||
try
|
||||
{
|
||||
std::rethrow_exception(dict_info.second.exception);
|
||||
}
|
||||
catch (...)
|
||||
{
|
||||
res_columns[i++]->insert(getCurrentExceptionMessage(false));
|
||||
}
|
||||
}
|
||||
res_columns[i++]->insert(static_cast<UInt64>(std::chrono::system_clock::to_time_t(load_result.loading_start_time)));
|
||||
res_columns[i++]->insert(std::chrono::duration_cast<std::chrono::duration<float>>(load_result.loading_duration).count());
|
||||
|
||||
if (load_result.exception)
|
||||
res_columns[i++]->insert(getExceptionMessage(load_result.exception, false));
|
||||
else
|
||||
res_columns[i++]->insertDefault();
|
||||
}
|
||||
|
|
|
|||
|
|
@ -2,9 +2,12 @@
|
|||
#include <DataTypes/DataTypeString.h>
|
||||
#include <DataTypes/DataTypesNumber.h>
|
||||
#include <DataTypes/DataTypeDateTime.h>
|
||||
#include <DataTypes/DataTypeEnum.h>
|
||||
#include <Interpreters/Context.h>
|
||||
#include <Interpreters/ExternalModels.h>
|
||||
#include <Interpreters/CatBoostModel.h>
|
||||
|
||||
|
||||
namespace DB
|
||||
{
|
||||
|
||||
|
|
@ -12,9 +15,12 @@ NamesAndTypesList StorageSystemModels::getNamesAndTypes()
|
|||
{
|
||||
return {
|
||||
{ "name", std::make_shared<DataTypeString>() },
|
||||
{ "status", std::make_shared<DataTypeEnum8>(ExternalLoader::getStatusEnumAllPossibleValues()) },
|
||||
{ "origin", std::make_shared<DataTypeString>() },
|
||||
{ "type", std::make_shared<DataTypeString>() },
|
||||
{ "creation_time", std::make_shared<DataTypeDateTime>() },
|
||||
{ "loading_start_time", std::make_shared<DataTypeDateTime>() },
|
||||
{ "loading_duration", std::make_shared<DataTypeFloat32>() },
|
||||
//{ "creation_time", std::make_shared<DataTypeDateTime>() },
|
||||
{ "last_exception", std::make_shared<DataTypeString>() },
|
||||
};
|
||||
}
|
||||
|
|
@ -22,40 +28,31 @@ NamesAndTypesList StorageSystemModels::getNamesAndTypes()
|
|||
void StorageSystemModels::fillData(MutableColumns & res_columns, const Context & context, const SelectQueryInfo &) const
|
||||
{
|
||||
const auto & external_models = context.getExternalModels();
|
||||
auto objects_map = external_models.getObjectsMap();
|
||||
const auto & models = objects_map.get();
|
||||
auto load_results = external_models.getCurrentLoadResults();
|
||||
|
||||
for (const auto & model_info : models)
|
||||
for (const auto & [name, load_result] : load_results)
|
||||
{
|
||||
res_columns[0]->insert(model_info.first);
|
||||
res_columns[1]->insert(model_info.second.origin);
|
||||
res_columns[0]->insert(name);
|
||||
res_columns[1]->insert(static_cast<Int8>(load_result.status));
|
||||
res_columns[2]->insert(load_result.origin);
|
||||
|
||||
if (model_info.second.loadable)
|
||||
if (load_result.object)
|
||||
{
|
||||
const auto model_ptr = std::static_pointer_cast<IModel>(model_info.second.loadable);
|
||||
|
||||
res_columns[2]->insert(model_ptr->getTypeName());
|
||||
res_columns[3]->insert(static_cast<UInt64>(std::chrono::system_clock::to_time_t(model_ptr->getCreationTime())));
|
||||
const auto model_ptr = std::static_pointer_cast<const IModel>(load_result.object);
|
||||
res_columns[3]->insert(model_ptr->getTypeName());
|
||||
}
|
||||
else
|
||||
{
|
||||
res_columns[2]->insertDefault();
|
||||
res_columns[3]->insertDefault();
|
||||
}
|
||||
|
||||
if (model_info.second.exception)
|
||||
{
|
||||
try
|
||||
{
|
||||
std::rethrow_exception(model_info.second.exception);
|
||||
}
|
||||
catch (...)
|
||||
{
|
||||
res_columns[4]->insert(getCurrentExceptionMessage(false));
|
||||
}
|
||||
}
|
||||
res_columns[4]->insert(static_cast<UInt64>(std::chrono::system_clock::to_time_t(load_result.loading_start_time)));
|
||||
res_columns[5]->insert(std::chrono::duration_cast<std::chrono::duration<float>>(load_result.loading_duration).count());
|
||||
|
||||
if (load_result.exception)
|
||||
res_columns[6]->insert(getExceptionMessage(load_result.exception, false));
|
||||
else
|
||||
res_columns[4]->insertDefault();
|
||||
res_columns[6]->insertDefault();
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -27,23 +27,16 @@ namespace ErrorCodes
|
|||
static NamesAndTypesList chooseColumns(const String & source_database, const String & table_name_regexp_, const Context & context)
|
||||
{
|
||||
OptimizedRegularExpression table_name_regexp(table_name_regexp_);
|
||||
auto table_name_match = [&](const String & table_name) { return table_name_regexp.match(table_name); };
|
||||
|
||||
StoragePtr any_table;
|
||||
|
||||
{
|
||||
auto database = context.getDatabase(source_database);
|
||||
auto iterator = database->getIterator(context);
|
||||
auto iterator = database->getIterator(context, table_name_match);
|
||||
|
||||
while (iterator->isValid())
|
||||
{
|
||||
if (table_name_regexp.match(iterator->name()))
|
||||
{
|
||||
any_table = iterator->table();
|
||||
break;
|
||||
}
|
||||
|
||||
iterator->next();
|
||||
}
|
||||
if (iterator->isValid())
|
||||
any_table = iterator->table();
|
||||
}
|
||||
|
||||
if (!any_table)
|
||||
|
|
|
|||
Loading…
Reference in New Issue