dbms: fixed global_in inside aggregate function. [METR-10974]

dbms/include/DB/Interpreters/ExpressionAnalyzer.h

@@ -21,24 +21,24 @@ namespace DB
 class ExpressionAnalyzer : private boost::noncopyable
 {
 public:
-	ExpressionAnalyzer(const ASTPtr & ast_, const Context & context_, size_t subquery_depth_ = 0)
+	ExpressionAnalyzer(const ASTPtr & ast_, const Context & context_, size_t subquery_depth_ = 0, bool do_global_ = false)
 		: ast(ast_), context(context_), settings(context.getSettings()),
-		subquery_depth(subquery_depth_), columns(context.getColumns()), storage(getTable())
+		subquery_depth(subquery_depth_), columns(context.getColumns()), storage(getTable()), do_global(do_global_)
 	{
 		init();
 	}
 
-	ExpressionAnalyzer(const ASTPtr & ast_, const Context & context_, StoragePtr storage_, size_t subquery_depth_ = 0)
+	ExpressionAnalyzer(const ASTPtr & ast_, const Context & context_, StoragePtr storage_, size_t subquery_depth_ = 0, bool do_global_ = false)
 		: ast(ast_), context(context_), settings(context.getSettings()),
-		subquery_depth(subquery_depth_), columns(context.getColumns()), storage(storage_ ? storage_ : getTable())
+		subquery_depth(subquery_depth_), columns(context.getColumns()), storage(storage_ ? storage_ : getTable()), do_global(do_global_)
 	{
 		init();
 	}
 
 	/// columns - список известных столбцов (которых можно достать из таблицы).
-	ExpressionAnalyzer(const ASTPtr & ast_, const Context & context_, const NamesAndTypesList & columns_, size_t subquery_depth_ = 0)
+	ExpressionAnalyzer(const ASTPtr & ast_, const Context & context_, const NamesAndTypesList & columns_, size_t subquery_depth_ = 0, bool do_global_ = false)
 		: ast(ast_), context(context_), settings(context.getSettings()),
-		subquery_depth(subquery_depth_), columns(columns_), storage(getTable())
+		subquery_depth(subquery_depth_), columns(columns_), storage(getTable()), do_global(do_global_)
 	{
 		init();
 	}
@@ -101,6 +101,7 @@ public:
 
 	/// Все новые временные таблицы, полученные при выполнении подзапросов GLOBAL IN.
 	Tables external_tables;
+	std::map<String, BlockInputStreamPtr> external_data;
 
 	/// Создаем какие сможем Set из секции In для использования индекса по ним
 	void makeSetsForIndex();
@@ -131,6 +132,7 @@ private:
 	AggregateDescriptions aggregate_descriptions;
 
 	std::map<std::string, SetPtr> sets_with_subqueries;
+	std::map<std::string, SetPtr> global_sets_with_subqueries;
 	typedef std::map<String, ASTPtr> Aliases;
 	Aliases aliases;
 	
@@ -144,6 +146,9 @@ private:
 	/// Для секции ARRAY JOIN отображение из алиаса в полное столбца
 	/// Например, для ARRAY JOIN [1,2] AS b сюда попадет "b"->"array(1,2)".
 	NameToNameMap array_join_alias_to_name;
+
+	/// Вычислять ли результат глобальных селектов при анализировании запроса.
+	bool do_global;
 	
 	/** Для getActionsImpl.
 	  * Стек из ExpressionActions, соответствующих вложенным лямбда-выражениям.
@@ -270,8 +275,7 @@ private:
 
 	/// Превратить перечисление значений или подзапрос в ASTSet. node - функция in или notIn.
 	void makeSet(ASTFunction * node, const Block & sample_block);
-	/// Выполнить подзапрос в секции GLOBAL IN и запомнить результат во временную таблицу типа memory
-	/// Все новые временные таблицы хранятся в переменной external_tables
+	/// Запустить подзапрос в секции GLOBAL IN, создать временную таблицу типа memory и запомнить эту пару в переменной external_tables
 	void addExternalStorage(ASTFunction * node, size_t & name_id);
 
 	void getArrayJoinedColumns();

dbms/src/Interpreters/ExpressionAnalyzer.cpp

@@ -81,6 +81,10 @@ void ExpressionAnalyzer::init()
 	select_query = dynamic_cast<ASTSelectQuery *>(&*ast);
 	has_aggregation = false;
 
+	/// Выполняем все GLOBAL IN подзапросы, результаты будут сохранены в query_analyzer->external_tables
+	if (do_global)
+		processGlobalOperations();
+
 	createAliasesDict(ast); /// Если есть агрегатные функции, присвоит has_aggregation=true.
 	normalizeTree();
 
@@ -88,6 +92,12 @@ void ExpressionAnalyzer::init()
 
 	removeUnusedColumns();
 
+	if (do_global)
+	{
+		for (auto & it : external_data)
+			copyData(*it.second, *external_tables[it.first]->write(ASTPtr()));
+	}
+
 	/// Найдем агрегатные функции.
 	if (select_query && (select_query->group_expression_list || select_query->having_expression))
 		has_aggregation = true;
@@ -637,13 +647,13 @@ void ExpressionAnalyzer::addExternalStorage(ASTFunction * node, size_t & name_id
 		String external_table_name = "_data" + toString(name_id++);
 		external_storage = StorageMemory::create(external_table_name, columns);
 		BlockOutputStreamPtr output = external_storage->write(ASTPtr());
-		copyData(*interpreter.execute(), *output);
 
 		ASTIdentifier * ast_ident = new ASTIdentifier();
 		ast_ident->kind = ASTIdentifier::Table;
 		ast_ident->name = external_storage->getTableName();
 		arg = ast_ident;
 		external_tables[external_table_name] = external_storage;
+		external_data[external_table_name] = interpreter.execute();
 	}
 	else
 		throw Exception("GLOBAL [NOT] IN supports only SELECT data.", ErrorCodes::BAD_ARGUMENTS);
@@ -1243,6 +1253,7 @@ void ExpressionAnalyzer::addMultipleArrayJoinAction(ExpressionActions & actions)
 
 void ExpressionAnalyzer::processGlobalOperations()
 {
+
 	std::vector<ASTPtr> global_nodes;
 	findGlobalFunctions(ast, global_nodes);
 

dbms/src/Interpreters/InterpreterSelectQuery.cpp

@@ -76,15 +76,12 @@ void InterpreterSelectQuery::init(BlockInputStreamPtr input_, const NamesAndType
 	if (context.getColumns().empty())
 		throw Exception("There are no available columns", ErrorCodes::THERE_IS_NO_COLUMN);
 
-	query_analyzer = new ExpressionAnalyzer(query_ptr, context, storage, subquery_depth);
-
-	/// Выполняем все GLOBAL IN подзапросы, результаты будут сохранены в query_analyzer->external_tables
-	query_analyzer->processGlobalOperations();
+	query_analyzer = new ExpressionAnalyzer(query_ptr, context, storage, subquery_depth, true);
 
 	/// Сохраняем в query context новые временные таблицы
 	for (auto & it : query_analyzer->external_tables)
 		if (!(context.tryGetExternalTable(it.first)))
-		context.addExternalTable(it.first, it.second);
+			context.addExternalTable(it.first, it.second);
 
 	if (input_)
 		streams.push_back(input_);