dbms: improved performance with very small INSERTs [#METR-12715].

dbms/include/DB/Common/ProfileEvents.h

@@ -8,53 +8,57 @@
   */
 
 #define APPLY_FOR_EVENTS(M) \
-	M(Query, 							"Queries") \
-	M(SelectQuery, 						"Select queries") \
-	M(InsertQuery, 						"Insert queries") \
-	M(FileOpen, 						"File opens") \
-	M(Seek, 							"Seeks") \
-	M(ReadBufferFromFileDescriptorRead, "ReadBufferFromFileDescriptor reads") \
-	M(ReadCompressedBytes, 				"Read compressed bytes") \
-	M(CompressedReadBufferBlocks, 		"Read decompressed blocks") \
-	M(CompressedReadBufferBytes, 		"Read decompressed bytes") \
-	M(UncompressedCacheHits, 			"Uncompressed cache hits") \
-	M(UncompressedCacheMisses, 			"Uncompressed cache misses") \
-	M(UncompressedCacheWeightLost,		"Uncompressed cache weight lost") \
-	M(IOBufferAllocs, 					"IO buffers allocations") \
-	M(IOBufferAllocBytes, 				"IO buffers allocated bytes") \
-	M(ArenaAllocChunks, 				"Arena allocated chunks") \
-	M(ArenaAllocBytes, 					"Arena allocated bytes") \
-	M(FunctionExecute, 					"Function executes") \
-	M(MarkCacheHits, 					"Mark cache hits") \
-	M(MarkCacheMisses, 					"Mark cache misses") \
+	M(Query) \
+	M(SelectQuery) \
+	M(InsertQuery) \
+	M(FileOpen) \
+	M(Seek) \
+	M(ReadBufferFromFileDescriptorRead) \
+	M(ReadCompressedBytes) \
+	M(CompressedReadBufferBlocks) \
+	M(CompressedReadBufferBytes) \
+	M(UncompressedCacheHits) \
+	M(UncompressedCacheMisses) \
+	M(UncompressedCacheWeightLost) \
+	M(IOBufferAllocs) \
+	M(IOBufferAllocBytes) \
+	M(ArenaAllocChunks) \
+	M(ArenaAllocBytes) \
+	M(FunctionExecute) \
+	M(MarkCacheHits) \
+	M(MarkCacheMisses) \
 	\
-	M(ReplicatedPartFetches, 			"Replicated part fetches") \
-	M(ReplicatedPartFailedFetches,		"Replicated part fetches failed") \
-	M(ObsoleteReplicatedParts,			"Replicated parts rendered obsolete by fetches") \
-	M(ReplicatedPartMerges,				"Replicated part merges") \
-	M(ReplicatedPartFetchesOfMerged,	"Replicated part merges replaced with fetches") \
-	M(ReplicatedPartChecks,				"Replicated part checks") \
-	M(ReplicatedPartChecksFailed,		"Replicated part checks failed") \
+	M(ReplicatedPartFetches) \
+	M(ReplicatedPartFailedFetches) \
+	M(ObsoleteReplicatedParts) \
+	M(ReplicatedPartMerges) \
+	M(ReplicatedPartFetchesOfMerged) \
+	M(ReplicatedPartChecks) \
+	M(ReplicatedPartChecksFailed) \
 	\
-	M(ZooKeeperInit,					"ZooKeeper session init") \
-	M(ZooKeeperTransactions,			"ZooKeeper transactions all types") \
-	M(ZooKeeperGetChildren,				"ZooKeeper get children") \
-	M(ZooKeeperCreate,					"ZooKeeper create") \
-	M(ZooKeeperRemove,					"ZooKeeper remove") \
-	M(ZooKeeperExists,					"ZooKeeper exists") \
-	M(ZooKeeperGet,						"ZooKeeper get") \
-	M(ZooKeeperSet,						"ZooKeeper set") \
-	M(ZooKeeperMulti,					"ZooKeeper multi") \
-	M(ZooKeeperExceptions,				"ZooKeeper exceptions") \
+	M(DelayedInserts) \
+	M(RejectedInserts) \
+	M(DelayedInsertsMilliseconds) \
 	\
-	M(END, "")
+	M(ZooKeeperInit) \
+	M(ZooKeeperTransactions) \
+	M(ZooKeeperGetChildren) \
+	M(ZooKeeperCreate) \
+	M(ZooKeeperRemove) \
+	M(ZooKeeperExists) \
+	M(ZooKeeperGet) \
+	M(ZooKeeperSet) \
+	M(ZooKeeperMulti) \
+	M(ZooKeeperExceptions) \
+	\
+	M(END)
 
 namespace ProfileEvents
 {
 	/// Виды событий.
 	enum Event
 	{
-	#define M(NAME, DESCRIPTION) NAME,
+	#define M(NAME) NAME,
 		APPLY_FOR_EVENTS(M)
 	#undef M
 	};
@@ -65,7 +69,7 @@ namespace ProfileEvents
 	{
 		static const char * descriptions[] =
 		{
-		#define M(NAME, DESCRIPTION) DESCRIPTION,
+		#define M(NAME) #NAME,
 			APPLY_FOR_EVENTS(M)
 		#undef M
 		};

dbms/include/DB/Storages/MergeTree/AbandonableLockInZooKeeper.h

@@ -13,8 +13,6 @@ namespace DB
   * unlock() разблокирует ее (UNLOCKED).
   * При вызове деструктора или завершении сессии в ZooKeeper, переходит в состояние ABANDONED.
   *  (В том числе при падении программы).
-  *
-  * Из состояния ABANDONED не может перейти ни в какое другое состояние.
   */
 class AbandonableLockInZooKeeper : private boost::noncopyable
 {

dbms/include/DB/Storages/MergeTree/MergeTreeData.h

@@ -66,6 +66,8 @@ namespace DB
 
 struct MergeTreeSettings
 {
+	/** Настройки слияний. */
+
 	/// Опеределяет, насколько разбалансированные объединения мы готовы делать.
 	/// Чем больше, тем более разбалансированные. Желательно, чтобы было больше, чем 1/max_parts_to_merge_at_once.
 	double size_ratio_coefficient_to_merge_parts = 0.25;
@@ -74,6 +76,11 @@ struct MergeTreeSettings
 	/// Трудоемкость выбора кусков O(N * max_parts_to_merge_at_once).
 	size_t max_parts_to_merge_at_once = 10;
 
+	/// Но пока суммарный размер кусков слишком маленький (меньше такого количества байт), можно сливать и больше кусков за раз.
+	/// Это сделано, чтобы быстрее сливать очень уж маленькие куски, которых может быстро накопиться много.
+	size_t merge_more_parts_if_sum_bytes_is_less_than = 100 * 1024 * 1024;
+	size_t max_parts_to_merge_at_once_if_small = 100;
+
 	/// Куски настолько большого размера объединять нельзя вообще.
 	size_t max_bytes_to_merge_parts = 25ul * 1024 * 1024 * 1024;
 
@@ -89,6 +96,11 @@ struct MergeTreeSettings
 	/// Если из одного файла читается хотя бы столько строк, чтение можно распараллелить.
 	size_t min_rows_for_concurrent_read = 20 * 8192;
 
+	/// Через сколько секунд удалять ненужные куски.
+	time_t old_parts_lifetime = 8 * 60;
+
+	/** Настройки чтения и работы с индексом. */
+
 	/// Можно пропускать чтение более чем стольки строк ценой одного seek по файлу.
 	size_t min_rows_for_seek = 5 * 8192;
 
@@ -99,8 +111,7 @@ struct MergeTreeSettings
 	/// (Чтобы большие запросы не вымывали кэш.)
 	size_t max_rows_to_use_cache = 1024 * 1024;
 
-	/// Через сколько секунд удалять ненужные куски.
-	time_t old_parts_lifetime = 8 * 60;
+	/** Настройки вставок. */
 
 	/// Если в таблице хотя бы столько активных кусков, искусственно замедлять вставки в таблицу.
 	size_t parts_to_delay_insert = 150;
@@ -109,6 +120,8 @@ struct MergeTreeSettings
 	/// Таким образом, скорость вставок автоматически замедлится примерно до скорости слияний.
 	double insert_delay_step = 1.1;
 
+	/** Настройки репликации. */
+
 	/// Для скольки последних блоков хранить хеши в ZooKeeper.
 	size_t replicated_deduplication_window = 100;
 

dbms/src/Server/Server.cpp

@@ -85,8 +85,6 @@ private:
 			prev_counters[i] = counter;
 
 			std::string key{ProfileEvents::getDescription(static_cast<ProfileEvents::Event>(i))};
-			std::replace(std::begin(key), std::end(key), ' ', '_');
-
 			key_vals.emplace_back(event_path_prefix + key, counter_increment);
 		}
 

dbms/src/Storages/MergeTree/MergeTreeData.cpp

@@ -853,7 +853,13 @@ void MergeTreeData::delayInsertIfNeeded(Poco::Event * until)
 		delay /= 1000;
 
 		if (delay > DBMS_MAX_DELAY_OF_INSERT)
+		{
+			ProfileEvents::increment(ProfileEvents::RejectedInserts);
 			throw Exception("Too much parts. Merges are processing significantly slower than inserts.", ErrorCodes::TOO_MUCH_PARTS);
+		}
+
+		ProfileEvents::increment(ProfileEvents::DelayedInserts);
+		ProfileEvents::increment(ProfileEvents::DelayedInsertsMilliseconds, delay * 1000);
 
 		LOG_INFO(log, "Delaying inserting block by "
 			<< std::fixed << std::setprecision(4) << delay << " sec. because there are " << parts_count << " parts");

dbms/src/Storages/MergeTree/MergeTreeDataMerger.cpp

@@ -20,22 +20,23 @@ static const double DISK_USAGE_COEFFICIENT_TO_SELECT = 1.6;
 static const double DISK_USAGE_COEFFICIENT_TO_RESERVE = 1.4;
 
 
-/// Выбираем отрезок из не более чем max_parts_to_merge_at_once кусков так, чтобы максимальный размер был меньше чем в max_size_ratio_to_merge_parts раз больше суммы остальных.
+/// Выбираем отрезок из не более чем max_parts_to_merge_at_once (или несколько больше, см merge_more_parts_if_sum_bytes_is_less_than)
+///  кусков так, чтобы максимальный размер был меньше чем в max_size_ratio_to_merge_parts раз больше суммы остальных.
 /// Это обеспечивает в худшем случае время O(n log n) на все слияния, независимо от выбора сливаемых кусков, порядка слияния и добавления.
-/// При max_parts_to_merge_at_once >= log(max_bytes_to_merge_parts)/log(max_size_ratio_to_merge_parts),
+/// При max_parts_to_merge_at_once >= log(max_bytes_to_merge_parts) / log(max_size_ratio_to_merge_parts),
 /// несложно доказать, что всегда будет что сливать, пока количество кусков больше
-/// log(max_bytes_to_merge_parts)/log(max_size_ratio_to_merge_parts)*(количество кусков размером больше max_bytes_to_merge_parts).
+/// log(max_bytes_to_merge_parts) / log(max_size_ratio_to_merge_parts) * (количество кусков размером больше max_bytes_to_merge_parts).
 /// Дальше эвристики.
 /// Будем выбирать максимальный по включению подходящий отрезок.
 /// Из всех таких выбираем отрезок с минимальным максимумом размера.
 /// Из всех таких выбираем отрезок с минимальным минимумом размера.
 /// Из всех таких выбираем отрезок с максимальной длиной.
 /// Дополнительно:
-/// 1) с 1:00 до 5:00 ограничение сверху на размер куска в основном потоке увеличивается в несколько раз
-/// 2) в зависимоти от возраста кусков меняется допустимая неравномерность при слиянии
-/// 3) Молодые куски крупного размера (примерно больше 1 Гб) можно сливать не меньше чем по три
-/// 4) Если в одном из потоков идет мердж крупных кусков, то во втором сливать только маленькие кусочки
-/// 5) С ростом логарифма суммарного размера кусочков в мердже увеличиваем требование сбалансированности
+/// 1) С 1:00 до 5:00 ограничение сверху на размер куска в основном потоке увеличивается в несколько раз.
+/// 2) В зависимоти от возраста кусков меняется допустимая неравномерность при слиянии.
+/// 3) Молодые куски крупного размера (примерно больше 1 ГБ) можно сливать не меньше чем по три.
+/// 4) Если в одном из потоков идет мердж крупных кусков, то во втором сливать только маленькие кусочки.
+/// 5) С ростом логарифма суммарного размера кусочков в мердже увеличиваем требование сбалансированности.
 
 bool MergeTreeDataMerger::selectPartsToMerge(MergeTreeData::DataPartsVector & parts, String & merged_name, size_t available_disk_space,
 	bool merge_anything_for_old_months, bool aggressive, bool only_small, const AllowedMergingPredicate & can_merge_callback)
@@ -128,7 +129,9 @@ bool MergeTreeDataMerger::selectPartsToMerge(MergeTreeData::DataPartsVector & pa
 
 		/// Правый конец отрезка.
 		MergeTreeData::DataParts::iterator jt = it;
-		for (; cur_len < static_cast<int>(data.settings.max_parts_to_merge_at_once);)
+		while (cur_len < static_cast<int>(data.settings.max_parts_to_merge_at_once)
+			|| (cur_len < static_cast<int>(data.settings.max_parts_to_merge_at_once_if_small)
+				&& cur_sum < data.settings.merge_more_parts_if_sum_bytes_is_less_than))
 		{
 			const MergeTreeData::DataPartPtr & prev_part = *jt;
 			++jt;