dbms: extract external dictionaries management to a separate type [#METR-13298]

dbms/include/DB/Dictionaries/DictionarySourceFactory.h

@@ -67,7 +67,7 @@ public:
 		}
 		else if ("mysql" == source_type)
 		{
-			return ext::make_unique<MySQLDictionarySource>(config, config_prefix + ".mysql", sample_block, context);
+			return ext::make_unique<MySQLDictionarySource>(config, config_prefix + ".mysql", sample_block);
 		}
 		else if ("clickhouse" == source_type)
 		{

dbms/include/DB/Dictionaries/MySQLDictionarySource.h

@@ -2,7 +2,6 @@
 
 #include <DB/Dictionaries/IDictionarySource.h>
 #include <DB/Dictionaries/MySQLBlockInputStream.h>
-#include <DB/Interpreters/Context.h>
 #include <statdaemons/ext/range.hpp>
 #include <mysqlxx/Pool.h>
 #include <Poco/Util/AbstractConfiguration.h>
@@ -18,9 +17,9 @@ class MySQLDictionarySource final : public IDictionarySource
 
 public:
 	MySQLDictionarySource(const Poco::Util::AbstractConfiguration & config, const std::string & config_prefix,
-		Block & sample_block, const Context & context)
+		Block & sample_block)
 		: table{config.getString(config_prefix + ".table")},
-		  sample_block{sample_block}, context(context),
+		  sample_block{sample_block},
 		  pool{config, config_prefix},
 		  load_all_query{composeLoadAllQuery(sample_block, table)},
 		  last_modification{getLastModification()}
@@ -29,7 +28,7 @@ public:
 	/// copy-constructor is provided in order to support cloneability
 	MySQLDictionarySource(const MySQLDictionarySource & other)
 		: table{other.table},
-		  sample_block{other.sample_block}, context(other.context),
+		  sample_block{other.sample_block},
 		  pool{other.pool},
 		  load_all_query{other.load_all_query}, last_modification{other.last_modification}
 	{}
@@ -107,7 +106,6 @@ private:
 
 	const std::string table;
 	Block sample_block;
-	const Context & context;
 	mutable mysqlxx::PoolWithFailover pool;
 	const std::string load_all_query;
 	mysqlxx::DateTime last_modification;

dbms/include/DB/Functions/FunctionsDictionaries.h

@@ -740,10 +740,10 @@ public:
 
 	static IFunction * create(const Context & context)
 	{
-		return new FunctionDictGetString{context.getDictionaries()};
+		return new FunctionDictGetString{context.getExternalDictionaries()};
 	};
 
-	FunctionDictGetString(const Dictionaries & dictionaries) : dictionaries(dictionaries) {}
+	FunctionDictGetString(const ExternalDictionaries & dictionaries) : dictionaries(dictionaries) {}
 
 	String getName() const override { return name; }
 
@@ -804,7 +804,7 @@ private:
 				ErrorCodes::ILLEGAL_COLUMN
 			};
 
-		auto dict = dictionaries.getExternalDictionary(dict_name_col->getData());
+		auto dict = dictionaries.getDictionary(dict_name_col->getData());
 		const auto dict_ptr = dict.get();
 
 		if (!executeDispatch<FlatDictionary>(block, arguments, result, dict_ptr) &&
@@ -889,7 +889,7 @@ private:
 		return false;
 	}
 
-	const Dictionaries & dictionaries;
+	const ExternalDictionaries & dictionaries;
 };
 
 
@@ -929,10 +929,10 @@ public:
 
 	static IFunction * create(const Context & context)
 	{
-		return new FunctionDictGet{context.getDictionaries()};
+		return new FunctionDictGet{context.getExternalDictionaries()};
 	};
 
-	FunctionDictGet(const Dictionaries & dictionaries) : dictionaries(dictionaries) {}
+	FunctionDictGet(const ExternalDictionaries & dictionaries) : dictionaries(dictionaries) {}
 
 	String getName() const override { return name; }
 
@@ -993,7 +993,7 @@ private:
 				ErrorCodes::ILLEGAL_COLUMN
 			};
 
-		auto dict = dictionaries.getExternalDictionary(dict_name_col->getData());
+		auto dict = dictionaries.getDictionary(dict_name_col->getData());
 		const auto dict_ptr = dict.get();
 
 		if (!executeDispatch<FlatDictionary>(block, arguments, result, dict_ptr) &&
@@ -1080,7 +1080,7 @@ private:
 		return false;
 	}
 
-	const Dictionaries & dictionaries;
+	const ExternalDictionaries & dictionaries;
 };
 
 template <typename DataType>
@@ -1106,10 +1106,10 @@ public:
 
 	static IFunction * create(const Context & context)
 	{
-		return new FunctionDictGetHierarchy{context.getDictionaries()};
+		return new FunctionDictGetHierarchy{context.getExternalDictionaries()};
 	};
 
-	FunctionDictGetHierarchy(const Dictionaries & dictionaries) : dictionaries(dictionaries) {}
+	FunctionDictGetHierarchy(const ExternalDictionaries & dictionaries) : dictionaries(dictionaries) {}
 
 	String getName() const override { return name; }
 
@@ -1161,7 +1161,7 @@ private:
 				ErrorCodes::ILLEGAL_COLUMN
 			};
 
-		auto dict = dictionaries.getExternalDictionary(dict_name_col->getData());
+		auto dict = dictionaries.getDictionary(dict_name_col->getData());
 		const auto dict_ptr = dict.get();
 
 		if (!dict->hasHierarchy())
@@ -1259,7 +1259,7 @@ private:
 		return false;
 	}
 
-	const Dictionaries & dictionaries;
+	const ExternalDictionaries & dictionaries;
 };
 
 
@@ -1270,10 +1270,10 @@ public:
 
 	static IFunction * create(const Context & context)
 	{
-		return new FunctionDictIsIn{context.getDictionaries()};
+		return new FunctionDictIsIn{context.getExternalDictionaries()};
 	};
 
-	FunctionDictIsIn(const Dictionaries & dictionaries) : dictionaries(dictionaries) {}
+	FunctionDictIsIn(const ExternalDictionaries & dictionaries) : dictionaries(dictionaries) {}
 
 	String getName() const override { return name; }
 
@@ -1342,7 +1342,7 @@ private:
 				ErrorCodes::ILLEGAL_COLUMN
 			};
 
-		auto dict = dictionaries.getExternalDictionary(dict_name_col->getData());
+		auto dict = dictionaries.getDictionary(dict_name_col->getData());
 		const auto dict_ptr = dict.get();
 
 		if (!dict->hasHierarchy())
@@ -1499,7 +1499,7 @@ private:
 		return false;
 	}
 
-	const Dictionaries & dictionaries;
+	const ExternalDictionaries & dictionaries;
 };
 
 

dbms/include/DB/Interpreters/Context.h

@@ -23,6 +23,7 @@
 #include <DB/Interpreters/Users.h>
 #include <DB/Interpreters/Quota.h>
 #include <DB/Interpreters/Dictionaries.h>
+#include <DB/Interpreters/ExternalDictionaries.h>
 #include <DB/Interpreters/ProcessList.h>
 #include <DB/Interpreters/Cluster.h>
 #include <DB/Interpreters/InterserverIOHandler.h>
@@ -87,6 +88,7 @@ struct ContextShared
 	DataTypeFactory data_type_factory;						/// Типы данных.
 	FormatFactory format_factory;							/// Форматы.
 	mutable SharedPtr<Dictionaries> dictionaries;			/// Словари Метрики. Инициализируются лениво.
+	mutable SharedPtr<ExternalDictionaries> external_dictionaries;
 	Users users;											/// Известные пользователи.
 	Quotas quotas;											/// Известные квоты на использование ресурсов.
 	mutable UncompressedCachePtr uncompressed_cache;		/// Кэш разжатых блоков.
@@ -259,6 +261,7 @@ public:
 	const DataTypeFactory & getDataTypeFactory() const						{ return shared->data_type_factory; }
 	const FormatFactory & getFormatFactory() const							{ return shared->format_factory; }
 	const Dictionaries & getDictionaries() const;
+	const ExternalDictionaries & getExternalDictionaries() const;
 
 	InterserverIOHandler & getInterserverIOHandler()						{ return shared->interserver_io_handler; }
 

dbms/include/DB/Interpreters/Dictionaries.h

@@ -1,13 +1,5 @@
 #pragma once
 
-#include <mutex>
-#include <thread>
-#include <unordered_map>
-#include <chrono>
-#include <random>
-
-#include <Poco/SharedPtr.h>
-
 #include <Yandex/MultiVersion.h>
 #include <Yandex/logger_useful.h>
 #include <statdaemons/RegionsHierarchies.h>
@@ -18,10 +10,7 @@
 namespace DB
 {
 
-using Poco::SharedPtr;
-
 class Context;
-class IDictionary;
 
 /// Словари Метрики, которые могут использоваться в функциях.
 
@@ -31,22 +20,15 @@ private:
 	MultiVersion<RegionsHierarchies> regions_hierarchies;
 	MultiVersion<TechDataHierarchy> tech_data_hierarchy;
 	MultiVersion<RegionsNames> regions_names;
-	mutable std::mutex external_dictionaries_mutex;
-	std::unordered_map<std::string, std::shared_ptr<MultiVersion<IDictionary>>> external_dictionaries;
-	std::unordered_map<std::string, std::chrono::system_clock::time_point> update_times;
-	std::mt19937 rnd_engine;
 
-	Context & context;
 	/// Периодичность обновления справочников, в секундах.
 	int reload_period;
 
 	std::thread reloading_thread;
-	std::thread reloading_externals_thread;
-	Poco::Event destroy{false};
+	Poco::Event destroy;
 
 	Logger * log;
 
-	Poco::Timestamp dictionaries_last_modified{0};
 
 
 	void handleException() const
@@ -122,7 +104,6 @@ private:
 	}
 
 
-	void reloadExternals();
 
 	/// Обновляет каждые reload_period секунд.
 	void reloadPeriodically()
@@ -136,35 +117,19 @@ private:
 		}
 	}
 
-	void reloadExternalsPeriodically()
-	{
-		const auto check_period = 5 * 1000;
-		while (true)
-		{
-			if (destroy.tryWait(check_period))
-				return;
-
-			reloadExternals();
-		}
-	}
-
 public:
 	/// Справочники будут обновляться в отдельном потоке, каждые reload_period секунд.
-	Dictionaries(Context & context, int reload_period_ = 3600)
-		: context(context), reload_period(reload_period_),
-		log(&Logger::get("Dictionaries"))
+	Dictionaries(int reload_period_ = 3600)
+		: reload_period(reload_period_), log(&Logger::get("Dictionaries"))
 	{
 		reloadImpl();
-		reloadExternals();
 		reloading_thread = std::thread([this] { reloadPeriodically(); });
-		reloading_externals_thread = std::thread{&Dictionaries::reloadExternalsPeriodically, this};
 	}
 
 	~Dictionaries()
 	{
 		destroy.set();
 		reloading_thread.join();
-		reloading_externals_thread.join();
 	}
 
 	MultiVersion<RegionsHierarchies>::Version getRegionsHierarchies() const
@@ -181,19 +146,6 @@ public:
 	{
 		return regions_names.get();
 	}
-
-	MultiVersion<IDictionary>::Version getExternalDictionary(const std::string & name) const
-	{
-		const std::lock_guard<std::mutex> lock{external_dictionaries_mutex};
-		const auto it = external_dictionaries.find(name);
-		if (it == std::end(external_dictionaries))
-			throw Exception{
-				"No such dictionary: " + name,
-				ErrorCodes::BAD_ARGUMENTS
-			};
-
-		return it->second->get();
-	}
 };
 
 }

dbms/include/DB/Interpreters/ExternalDictionaries.h

@@ -0,0 +1,122 @@
+#pragma once
+
+#include <DB/Core/Exception.h>
+#include <DB/Core/ErrorCodes.h>
+#include <Yandex/MultiVersion.h>
+#include <Yandex/logger_useful.h>
+#include <Poco/Event.h>
+#include <time.h>
+#include <mutex>
+#include <thread>
+#include <unordered_map>
+#include <chrono>
+#include <random>
+#include <unistd.h>
+
+namespace DB
+{
+
+class Context;
+class IDictionary;
+
+/** Manages user-defined dictionaries.
+*	Monitors configuration file and automatically reloads dictionaries in a separate thread.
+*	The monitoring thread wakes up every @check_period_sec seconds and checks
+*	modification time of dictionaries' configuration file. If said time is greater than
+*	@config_last_modified, the dictionaries are created from scratch using configuration file,
+*	possibly overriding currently existing dictionaries with the same name (previous versions of
+*	overridden dictionaries will live as long as there are any users retaining them).
+*
+*	Apart from checking configuration file for modifications, each non-cached dictionary
+*	has a lifetime of its own and may be updated if it's source reports that it has been
+*	modified. 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 dictionary is never updated.
+*/
+class ExternalDictionaries
+{
+private:
+	static const auto check_period_sec = 5;
+
+	mutable std::mutex dictionaries_mutex;
+	std::unordered_map<std::string, std::shared_ptr<MultiVersion<IDictionary>>> dictionaries;
+	std::unordered_map<std::string, std::chrono::system_clock::time_point> update_times;
+	std::mt19937_64 rnd_engine{getSeed()};
+
+	Context & context;
+
+	std::thread reloading_thread;
+	Poco::Event destroy;
+
+	Logger * log;
+
+	Poco::Timestamp config_last_modified{0};
+
+	void handleException() const
+	{
+		try
+		{
+			throw;
+		}
+		catch (const Poco::Exception & e)
+		{
+			LOG_ERROR(log, "Cannot load exter dictionary! You must resolve this manually. " << e.displayText());
+			return;
+		}
+		catch (...)
+		{
+			LOG_ERROR(log, "Cannot load dictionary! You must resolve this manually.");
+			return;
+		}
+	}
+
+	void reloadImpl();
+
+	void reloadPeriodically()
+	{
+		while (true)
+		{
+			if (destroy.tryWait(check_period_sec * 1000))
+				return;
+
+			reloadImpl();
+		}
+	}
+
+	static std::uint64_t getSeed()
+	{
+		timespec ts;
+		clock_gettime(CLOCK_MONOTONIC, &ts);
+		return ts.tv_nsec ^ getpid();
+	}
+
+public:
+	/// Справочники будут обновляться в отдельном потоке, каждые reload_period секунд.
+	ExternalDictionaries(Context & context)
+		: context(context), log(&Logger::get("ExternalDictionaries"))
+	{
+		reloadImpl();
+		reloading_thread = std::thread{&ExternalDictionaries::reloadPeriodically, this};
+	}
+
+	~ExternalDictionaries()
+	{
+		destroy.set();
+		reloading_thread.join();
+	}
+
+	MultiVersion<IDictionary>::Version getDictionary(const std::string & name) const
+	{
+		const std::lock_guard<std::mutex> lock{dictionaries_mutex};
+		const auto it = dictionaries.find(name);
+		if (it == std::end(dictionaries))
+			throw Exception{
+				"No such dictionary: " + name,
+				ErrorCodes::BAD_ARGUMENTS
+			};
+
+		return it->second->get();
+	}
+};
+
+}

dbms/src/Interpreters/Context.cpp

@@ -495,13 +495,24 @@ const Dictionaries & Context::getDictionaries() const
 	Poco::ScopedLock<Poco::Mutex> lock(shared->mutex);
 
 	if (!shared->dictionaries)
+		shared->dictionaries = new Dictionaries;
+
+	return *shared->dictionaries;
+}
+
+
+const ExternalDictionaries & Context::getExternalDictionaries() const
+{
+	Poco::ScopedLock<Poco::Mutex> lock(shared->mutex);
+
+	if (!shared->external_dictionaries)
 	{
 		if (!this->global_context)
 			throw Exception("Logical error: there is no global context", ErrorCodes::LOGICAL_ERROR);
-		shared->dictionaries = new Dictionaries{ *this->global_context };
+		shared->external_dictionaries = new ExternalDictionaries{*this->global_context};
 	}
 
-	return *shared->dictionaries;
+	return *shared->external_dictionaries;
 }
 
 

dbms/src/Interpreters/ExternalDictionaries.cpp

@@ -1,8 +1,9 @@
-#include <DB/Interpreters/Dictionaries.h>
+#include <DB/Interpreters/ExternalDictionaries.h>
 #include <DB/Dictionaries/DictionaryFactory.h>
 #include <DB/Dictionaries/DictionaryStructure.h>
 #include <DB/Dictionaries/IDictionarySource.h>
 #include <statdaemons/ext/scope_guard.hpp>
+#include <Poco/Util/Application.h>
 
 namespace DB
 {
@@ -28,7 +29,7 @@ namespace
 	}
 }
 
-void Dictionaries::reloadExternals()
+void ExternalDictionaries::reloadImpl()
 {
 	const auto config_path = getDictionariesConfigPath(Poco::Util::Application::instance().config());
 	const Poco::File config_file{config_path};
@@ -40,10 +41,10 @@ void Dictionaries::reloadExternals()
 	else
 	{
 		const auto last_modified = config_file.getLastModified();
-		if (last_modified > dictionaries_last_modified)
+		if (last_modified > config_last_modified)
 		{
 			/// definitions of dictionaries may have changed, recreate all of them
-			dictionaries_last_modified = last_modified;
+			config_last_modified = last_modified;
 
 			const auto config = new Poco::Util::XMLConfiguration{config_path};
 			SCOPE_EXIT(
@@ -87,12 +88,12 @@ void Dictionaries::reloadExternals()
 						}
 					}
 
-					auto it = external_dictionaries.find(name);
+					auto it = dictionaries.find(name);
 					/// add new dictionary or update an existing version
-					if (it == std::end(external_dictionaries))
+					if (it == std::end(dictionaries))
 					{
-						const std::lock_guard<std::mutex> lock{external_dictionaries_mutex};
-						external_dictionaries.emplace(name, std::make_shared<MultiVersion<IDictionary>>(dict_ptr.release()));
+						const std::lock_guard<std::mutex> lock{dictionaries_mutex};
+						dictionaries.emplace(name, std::make_shared<MultiVersion<IDictionary>>(dict_ptr.release()));
 					}
 					else
 						it->second->set(dict_ptr.release());
@@ -106,7 +107,7 @@ void Dictionaries::reloadExternals()
 	}
 
 	/// periodic update
-	for (auto & dictionary : external_dictionaries)
+	for (auto & dictionary : dictionaries)
 	{
 		try
 		{