Merge

dbms/include/DB/Common/ProfileEvents.h

@@ -66,6 +66,9 @@
 	M(DistributedConnectionFailTry) \
 	M(DistributedConnectionFailAtAll) \
 	\
+	M(CompileAttempt) \
+	M(CompileSuccess) \
+	\
 	M(END)
 
 namespace ProfileEvents

dbms/include/DB/Interpreters/Cluster.h

@@ -63,6 +63,7 @@ public:
 		* </shard>
 		*/
 		Poco::Net::SocketAddress host_port;
+		String host_name;
 		String user;
 		String password;
 		UInt32 replica_num;

dbms/include/DB/Interpreters/Limits.h

@@ -61,7 +61,7 @@ struct Limits
 	M(SettingUInt64, max_subquery_depth, 100) \
 	M(SettingUInt64, max_pipeline_depth, 1000) \
 	M(SettingUInt64, max_ast_depth, 1000)		/** Проверяются не во время парсинга, */ \
-	M(SettingUInt64, max_ast_elements, 10000)	/**  а уже после парсинга запроса. */ \
+	M(SettingUInt64, max_ast_elements, 50000)	/**  а уже после парсинга запроса. */ \
 	\
 	/** 0 - можно всё. 1 - только запросы на чтение. 2 - только запросы на чтение, а также изменение настроек, кроме настройки readonly. */ \
 	M(SettingUInt64, readonly, 0) \

dbms/src/Interpreters/Cluster.cpp

@@ -13,8 +13,9 @@ Cluster::Address::Address(const String & config_prefix)
 {
 	auto & config = Poco::Util::Application::instance().config();
 
+	host_name = config.getString(config_prefix + ".host");
 	host_port = Poco::Net::SocketAddress(
-		config.getString(config_prefix + ".host"),
+		host_name,
 		config.getInt(config_prefix + ".port")
 	);
 
@@ -68,7 +69,7 @@ Cluster::Cluster(const Settings & settings, const DataTypeFactory & data_type_fa
 
 	const auto & config_prefix = cluster_name + ".";
 
-	UInt32 current_shard_num = 0;
+	UInt32 current_shard_num = 1;
 
 	for (auto it = config_keys.begin(); it != config_keys.end(); ++it)
 	{
@@ -80,7 +81,7 @@ Cluster::Cluster(const Settings & settings, const DataTypeFactory & data_type_fa
 				continue;
 
 			addresses.emplace_back(prefix);
-			addresses.back().replica_num = 0;
+			addresses.back().replica_num = 1;
 
 			slot_to_shard.insert(std::end(slot_to_shard), weight, shard_info_vec.size());
 			if (const auto is_local = isLocal(addresses.back()))
@@ -95,7 +96,7 @@ Cluster::Cluster(const Settings & settings, const DataTypeFactory & data_type_fa
 
 			addresses_with_failover.emplace_back();
 			Addresses & replica_addresses = addresses_with_failover.back();
-			UInt32 current_replica_num = 0;
+			UInt32 current_replica_num = 1;
 
 			const auto & partial_prefix = config_prefix + *it + ".";
 			const auto weight = config.getInt(partial_prefix + ".weight", 1);

dbms/src/Interpreters/Compiler.cpp

@@ -191,6 +191,8 @@ void Compiler::compile(
 	CodeGenerator get_code,
 	ReadyCallback on_ready)
 {
+	ProfileEvents::increment(ProfileEvents::CompileAttempt);
+
 	std::string prefix = path + "/" + file_name;
 	std::string cpp_file_path = prefix + ".cpp";
 	std::string so_file_path = prefix + ".so";
@@ -257,6 +259,7 @@ void Compiler::compile(
 	}
 
 	LOG_INFO(log, "Compiled code " << file_name);
+	ProfileEvents::increment(ProfileEvents::CompileSuccess);
 
 	on_ready(lib);
 }

dbms/src/Interpreters/InterpreterSelectQuery.cpp

@@ -502,7 +502,7 @@ void InterpreterSelectQuery::executeSingleQuery()
 
 		if (second_stage)
 		{
-			bool need_second_distinct_pass = true;
+			bool need_second_distinct_pass = query.distinct;
 
 			if (need_aggregate)
 			{
@@ -518,7 +518,7 @@ void InterpreterSelectQuery::executeSingleQuery()
 				executeExpression(streams, before_order_and_select);
 				executeDistinct(streams, true, selected_columns);
 
-				need_second_distinct_pass = streams.size() > 1;
+				need_second_distinct_pass = query.distinct && (streams.size() > 1);
 			}
 			else if (query.group_by_with_totals && !aggregate_final)
 			{
@@ -551,6 +551,9 @@ void InterpreterSelectQuery::executeSingleQuery()
 			if (query.limit_length && streams.size() > 1 && !query.distinct)
 				executePreLimit(streams);
 
+			if (need_second_distinct_pass)
+				union_within_single_query = true;
+
 			if (union_within_single_query)
 				executeUnion(streams);
 

dbms/src/Storages/StorageMergeTree.cpp

@@ -262,7 +262,8 @@ void StorageMergeTree::dropPartition(const Field & partition, bool detach, bool
 	/// Просит завершить мерджи и не позволяет им начаться.
 	/// Это защищает от "оживания" данных за удалённую партицию после завершения мерджа.
 	const MergeTreeMergeBlocker merge_blocker{merger};
-	auto structure_lock = lockStructure(true);
+	/// Дожидается завершения мерджей и не даёт начаться новым.
+	auto lock = lockForAlter();
 
 	DayNum_t month = MergeTreeData::getMonthDayNum(partition);
 

dbms/src/Storages/StorageSystemClusters.cpp

@@ -61,11 +61,8 @@ BlockInputStreams StorageSystemClusters::read(
 		shard_weight_column->insert(static_cast<UInt64>(shard_info.weight));
 		replica_num_column->insert(static_cast<UInt64>(address.replica_num));
 
-		const std::string & source = address.host_port.host().toString();
-		const auto host_entry = Poco::Net::DNS::resolve(source);
-		host_name_column->insert(host_entry.name());
-		host_address_column->insert(host_entry.addresses()[0].toString());
-
+		host_name_column->insert(address.host_name);
+		host_address_column->insert(address.host_port.host().toString());
 		port_column->insert(static_cast<UInt64>(address.host_port.port()));
 		user_column->insert(address.user);
 	};

dbms/tests/queries/0_stateless/00154_distributed_with_distinct.reference

@@ -0,0 +1,10 @@
+0
+1
+2
+3
+4
+5
+6
+7
+8
+9

dbms/tests/queries/0_stateless/00154_distributed_with_distinct.sql

@@ -0,0 +1 @@
+SELECT DISTINCT number FROM remote('127.0.0.{1,2}', system.numbers) LIMIT 10