Merge

dbms/include/DB/DataStreams/MergingSortedBlockInputStream.h

@@ -87,7 +87,7 @@ protected:
 
 	/// Эти методы используются в Collapsing/Summing SortedBlockInputStream-ах.
 
-	/// Сохранить строчку, на которую указывает cursor в row.
+	/// Сохранить строчку, на которую указывает cursor, в row.
 	template<class TSortCursor>
 	void setRow(Row & row, TSortCursor & cursor)
 	{

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

@@ -16,6 +16,15 @@ public:
 		auto part_blocks = storage.writer.splitBlockIntoParts(block);
 		for (auto & current_block : part_blocks)
 		{
+			size_t parts_count = storage.data.getDataPartsCount();
+			if (parts_count > storage.data.settings.parts_to_delay_insert)
+			{
+				double delay = std::pow(storage.data.settings.insert_delay_step, parts_count - storage.data.settings.parts_to_delay_insert);
+				delay /= 1000;
+				LOG_INFO(storage.log, "Delaying inserting block by " << delay << "s because there are " << parts_count << " parts");
+				std::this_thread::sleep_for(std::chrono::duration<double>(delay));
+			}
+
 			UInt64 temp_index = storage.increment.get();
 			MergeTreeData::MutableDataPartPtr part = storage.writer.writeTempPart(current_block, temp_index);
 			storage.data.renameTempPartAndAdd(part, &storage.increment);

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

@@ -100,6 +100,13 @@ struct MergeTreeSettings
 
 	/// Через сколько секунд удалять ненужные куски.
 	time_t old_parts_lifetime = 5 * 60;
+
+	/// Если в таблице хотя бы столько активных кусков, искусственно замедлять вставки в таблицу.
+	size_t parts_to_delay_insert = 150;
+
+	/// Если в таблице parts_to_delay_insert + k кусков, спать insert_delay_step^k миллисекунд перед вставкой каждого блока.
+	/// Таким образом, скорость вставок автоматически замедлится примерно до скорости слияний.
+	double insert_delay_step = 1.1;
 };
 
 class MergeTreeData : public ITableDeclaration
@@ -344,6 +351,7 @@ public:
 	  */
 	DataParts getDataParts();
 	DataParts getAllDataParts();
+	size_t getDataPartsCount();
 
 	/** Возвращает кусок с указанным именем или кусок, покрывающий его. Если такого нет, возвращает nullptr.
 	  * Если including_inactive, просматриваются также неактивные куски (all_data_parts).

dbms/src/Storages/MergeTree/MergeTreeData.cpp

@@ -755,6 +755,13 @@ MergeTreeData::DataParts MergeTreeData::getAllDataParts()
 	return all_data_parts;
 }
 
+size_t MergeTreeData::getDataPartsCount()
+{
+	Poco::ScopedLock<Poco::FastMutex> lock(data_parts_mutex);
+
+	return data_parts.size();
+}
+
 MergeTreeData::DataPartPtr MergeTreeData::getContainingPart(const String & part_name, bool including_inactive)
 {
 	MutableDataPartPtr tmp_part(new DataPart(*this));

dbms/src/Storages/tests/active_parts.py

@@ -0,0 +1,32 @@
+#!/usr/bin/python
+
+# Выводит список активных кусков - кусков, не покрытых никаким другим куском.
+# Использование: `ls /opt/clickhouse/data/merge/visits | active_parts.py`
+
+import sys
+import re
+
+parts = {}
+for s in sys.stdin.read().split():
+        m = re.match('^([0-9]{8})_([0-9]{8})_([0-9]+)_([0-9]+)_([0-9]+)$', s)
+        if m == None:
+                continue
+        m1 = m.group(1)
+        m2 = m.group(2)
+        i1 = int(m.group(3))
+        i2 = int(m.group(4))
+        l = int(m.group(5))
+        if m1 != m2:
+                raise Exception('not in single month: ' + s)
+        if m1 not in parts:
+                parts[m1] = []
+        parts[m1].append((i1, i2, l, s))
+
+for m in parts:
+        for x1, y1, l1, s1 in parts[m]:
+                for x2, y2, l2, s2 in parts[m]:
+                        if x1 >= x2 and y1 <= y2 and l1 < l2:
+                                break
+                else:
+                        print s1
+