dbms: prefer fetching part from replica, in some cases [#METR-22288].

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

@@ -30,13 +30,13 @@ struct MergeTreeSettings
 	size_t max_parts_to_merge_at_once_if_small = 100;
 
 	/// Куски настолько большого размера объединять нельзя вообще.
-	size_t max_bytes_to_merge_parts = 10ul * 1024 * 1024 * 1024;
+	size_t max_bytes_to_merge_parts = 10ULL * 1024 * 1024 * 1024;
 
 	/// Не больше половины потоков одновременно могут выполнять слияния, в которых участвует хоть один кусок хотя бы такого размера.
 	size_t max_bytes_to_merge_parts_small = 250 * 1024 * 1024;
 
 	/// Куски настолько большого размера в сумме, объединять нельзя вообще.
-	size_t max_sum_bytes_to_merge_parts = 25ul * 1024 * 1024 * 1024;
+	size_t max_sum_bytes_to_merge_parts = 25ULL * 1024 * 1024 * 1024;
 
 	/// Во столько раз ночью увеличиваем коэффициент.
 	size_t merge_parts_at_night_inc = 10;
@@ -68,6 +68,13 @@ struct MergeTreeSettings
 	/// Не влияет на работу таблиц; используется только чтобы успеть посмотреть на лог в ZooKeeper глазами прежде, чем его очистят.
 	size_t replicated_logs_to_keep = 100;
 
+	/// After specified amount of time passed after replication log entry creation
+	///  and sum size of parts is greater than threshold,
+	///  prefer fetching merged part from replica instead of doing merge locally.
+	/// To speed up very long merges.
+	time_t prefer_fetch_merged_part_time_threshold = 3600;
+	size_t prefer_fetch_merged_part_size_threshold = 10ULL * 1024 * 1024 * 1024;
+
 	/// Настройки минимального количества битых данных, при котором отказываться автоматически их удалять.
 	size_t max_suspicious_broken_parts = 10;
 
@@ -123,6 +130,8 @@ struct MergeTreeSettings
 		SET_DOUBLE(insert_delay_step);
 		SET_SIZE_T(replicated_deduplication_window);
 		SET_SIZE_T(replicated_logs_to_keep);
+		SET_SIZE_T(prefer_fetch_merged_part_time_threshold);
+		SET_SIZE_T(prefer_fetch_merged_part_size_threshold);
 		SET_SIZE_T(max_suspicious_broken_parts);
 		SET_SIZE_T(max_files_to_modify_in_alter_columns);
 		SET_SIZE_T(max_files_to_remove_in_alter_columns);

dbms/src/Storages/StorageReplicatedMergeTree.cpp

@@ -1100,7 +1100,27 @@ bool StorageReplicatedMergeTree::executeLogEntry(const LogEntry & entry, Backgro
 			do_fetch = true;
 			LOG_DEBUG(log, "Don't have all parts for merge " << entry.new_part_name << "; will try to fetch it instead");
 		}
-		else
+		else if (entry.create_time + data.settings.prefer_fetch_merged_part_time_threshold <= time(nullptr))
+		{
+			/// If entry is old enough, and have enough size, and part are exists in any replica,
+			///  then prefer fetching of merged part from replica.
+
+			size_t sum_parts_size_in_bytes = 0;
+			for (const auto & part : parts)
+				sum_parts_size_in_bytes += part->size_in_bytes;
+
+			if (sum_parts_size_in_bytes >= data.settings.prefer_fetch_merged_part_size_threshold)
+			{
+				String replica = findReplicaHavingPart(entry.new_part_name, true);	/// NOTE excessive ZK requests for same data later, may remove.
+				if (!replica.empty())
+				{
+					do_fetch = true;
+					LOG_DEBUG(log, "Preffering to fetch " << entry.new_part_name << " from replica");
+				}
+			}
+		}
+
+		if (!do_fetch)
 		{
 			/// Если собираемся сливать большие куски, увеличим счетчик потоков, сливающих большие куски.
 			for (const auto & part : parts)