Merge

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

@@ -2,6 +2,7 @@
 
 #include <DB/Storages/StorageReplicatedMergeTree.h>
 #include <DB/Storages/MergeTree/AbandonableLockInZooKeeper.h>
+#include <Yandex/time2str.h>
 
 
 namespace DB
@@ -20,9 +21,13 @@ public:
 		{
 			++block_index;
 			String block_id = insert_id.empty() ? "" : insert_id + "__" + toString(block_index);
+			time_t min_date_time = DateLUTSingleton::instance().fromDayNum(DayNum_t(current_block.min_date));
+			String month_name = toString(Date2OrderedIdentifier(min_date_time) / 100);
+
+			storage.zookeeper.tryCreate(storage.zookeeper_path + "/block_numbers/" + month_name, "", zkutil::CreateMode::Persistent);
 
 			AbandonableLockInZooKeeper block_number_lock(
-				storage.zookeeper_path + "/block_numbers/block-",
+				storage.zookeeper_path + "/block_numbers/" + month_name + "/block-",
 				storage.zookeeper_path + "/temp", storage.zookeeper);
 
 			UInt64 part_number = block_number_lock.getNumber();

dbms/src/Storages/StorageReplicatedMergeTree.cpp

@@ -869,10 +869,13 @@ void StorageReplicatedMergeTree::mergeSelectingThread()
 				}
 			}
 
+			if (success)
+			{
 				/// Нужно загрузить новую запись в очередь перед тем, как в следующий раз выбирать куски для слияния.
 				///  (чтобы куски пометились как currently_merging).
 				pullLogsToQueue();
 
+				String month_name = parts[0]->name.substr(0, 6);
 				for (size_t i = 0; i + 1 < parts.size(); ++i)
 				{
 					/// Уберем больше не нужные отметки о несуществующих блоках.
@@ -881,12 +884,13 @@ void StorageReplicatedMergeTree::mergeSelectingThread()
 						String number_str = toString(number);
 						while (number_str.size() < 10)
 							number_str = '0' + number_str;
-					String path = zookeeper_path + "/block_numbers/block-" + number_str;
+						String path = zookeeper_path + "/block_numbers/" + month_name + "/block-" + number_str;
 
 						zookeeper.tryRemove(path);
 					}
 				}
 			}
+		}
 		catch (...)
 		{
 			tryLogCurrentException(__PRETTY_FUNCTION__);
@@ -929,13 +933,15 @@ bool StorageReplicatedMergeTree::canMergeParts(const MergeTreeData::DataPartPtr
 	if (currently_merging.count(left->name) || currently_merging.count(right->name))
 		return false;
 
+	String month_name = left->name.substr(0, 6);
+
 	/// Можно слить куски, если все номера между ними заброшены - не соответствуют никаким блокам.
 	for (UInt64 number = left->right + 1; number <= right->left - 1; ++number)
 	{
 		String number_str = toString(number);
 		while (number_str.size() < 10)
 			number_str = '0' + number_str;
-		String path = zookeeper_path + "/block_numbers/block-" + number_str;
+		String path = zookeeper_path + "/block_numbers/" + month_name + "/block-" + number_str;
 
 		if (AbandonableLockInZooKeeper::check(path, zookeeper) != AbandonableLockInZooKeeper::ABANDONED)
 		{

libs/libzkutil/include/zkutil/ZooKeeper.h

@@ -61,6 +61,7 @@ public:
 	  * При остальных ошибках бросает исключение.
 	  */
 	ReturnCode::type tryCreate(const std::string & path, const std::string & data, CreateMode::type mode, std::string & pathCreated);
+	ReturnCode::type tryCreate(const std::string & path, const std::string & data, CreateMode::type mode);
 
 	/** Удалить ноду, если ее версия равна version (если -1, подойдет любая версия).
 	  */

libs/libzkutil/src/ZooKeeper.cpp

@@ -195,6 +195,12 @@ ReturnCode::type ZooKeeper::tryCreate(const std::string & path, const std::strin
 	return code;
 }
 
+ReturnCode::type ZooKeeper::tryCreate(const std::string & path, const std::string & data, CreateMode::type mode)
+{
+	std::string path_created;
+	return tryCreate(path, data, mode, path_created);
+}
+
 void ZooKeeper::remove(const std::string & path, int32_t version)
 {
 	checkNotExpired();