dbms: fixed error with -State functions and WITH TOTALS [#METR-18727].

dbms/include/DB/AggregateFunctions/AggregateFunctionState.h

@@ -100,6 +100,7 @@ public:
 	/// Аггрегатная функция или состояние аггрегатной функции.
 	bool isState() const { return true; }
 
+	AggregateFunctionPtr getNestedFunction() const { return nested_func_owner; }
 };
 
 }

dbms/include/DB/Columns/ColumnAggregateFunction.h

@@ -103,30 +103,9 @@ public:
 		arenas.push_back(arena_);
 	}
 
-	ColumnPtr convertToValues() const
-	{
-		const IAggregateFunction * function = holder->func;
-		ColumnPtr res = function->getReturnType()->createColumn();
-
-		/** Если агрегатная функция возвращает нефинализированное состояние,
-		  * то надо просто скопировать указатели на него а также разделяемое владение аренами.
-		  */
-		if (typeid_cast<const ColumnAggregateFunction *>(res.get()))
-		{
-			ColumnAggregateFunction * res_ = new ColumnAggregateFunction(*this);
-			res = res_;
-			res_->getData().assign(getData().begin(), getData().end());
-			return res;
-		}
-
-		IColumn & column = *res;
-		res->reserve(getData().size());
-
-		for (auto val : getData())
-			function->insertResultInto(val, column);
-
-		return res;
-	}
+	/** Преобразовать столбец состояний агрегатной функции в столбец с готовыми значениями результатов.
+	  */
+	ColumnPtr convertToValues() const;
 
 	std::string getName() const override { return "ColumnAggregateFunction"; }
 

dbms/src/Columns/ColumnAggregateFunction.cpp

@@ -0,0 +1,62 @@
+#include <DB/AggregateFunctions/AggregateFunctionState.h>
+#include <DB/Columns/ColumnAggregateFunction.h>
+
+
+namespace DB
+{
+
+ColumnPtr ColumnAggregateFunction::convertToValues() const
+{
+	const IAggregateFunction * function = holder->func;
+	ColumnPtr res = function->getReturnType()->createColumn();
+
+	/** Если агрегатная функция возвращает нефинализированное состояние,
+		* то надо просто скопировать указатели на него а также разделяемое владение данными.
+		*
+		* Также заменяем агрегатную функцию на вложенную.
+		* То есть, если этот столбец - состояния агрегатной функции aggState,
+		* то мы возвращаем такой же столбец, но с состояниями агрегатной функции agg.
+		* Это одни и те же состояния, меняем только функцию, которой они соответствуют.
+		*
+		* Это довольно сложно для понимания.
+		* Пример, когда такое происходит:
+		*
+		* SELECT k, finalizeAggregation(quantileTimingState(0.5)(x)) FROM ... GROUP BY k WITH TOTALS
+		*
+		* Здесь вычисляется агрегатная функция quantileTimingState.
+		* Её тип возвращаемого значения:
+		*  AggregateFunction(quantileTiming(0.5), UInt64).
+		* Из-за наличия WITH TOTALS, при агрегации будут сохранены состояния этой агрегатной функции
+		*  в столбце ColumnAggregateFunction, имеющего тип
+		*  AggregateFunction(quantileTimingState(0.5), UInt64).
+		* Затем в TotalsHavingBlockInputStream у него будет вызван метод convertToValues,
+		*  чтобы получить "готовые" значения.
+		* Но он всего лишь преобразует столбец типа
+		*   AggregateFunction(quantileTimingState(0.5), UInt64)
+		* в AggregateFunction(quantileTiming(0.5), UInt64)
+		* - в такие же состояния.
+		*
+		* Затем будет вычислена функция finalizeAggregation, которая позовёт convertToValues уже от результата.
+		* И это преобразует столбец типа
+		*   AggregateFunction(quantileTiming(0.5), UInt64)
+		* в UInt16 - уже готовый результат работы quantileTiming.
+		*/
+	if (const AggregateFunctionState * function_state = typeid_cast<const AggregateFunctionState *>(function))
+	{
+		ColumnAggregateFunction * res_ = new ColumnAggregateFunction(*this);
+		res = res_;
+		res_->set(function_state->getNestedFunction());
+		res_->getData().assign(getData().begin(), getData().end());
+		return res;
+	}
+
+	IColumn & column = *res;
+	res->reserve(getData().size());
+
+	for (auto val : getData())
+		function->insertResultInto(val, column);
+
+	return res;
+}
+
+}

dbms/tests/queries/0_stateless/00271_agg_state_and_totals.reference

@@ -0,0 +1,6 @@
+0	[15008]
+1	[15008]
+2	[15008]
+3	[5001]
+
+0	[13341]

dbms/tests/queries/0_stateless/00271_agg_state_and_totals.sql

@@ -0,0 +1 @@
+SELECT k, finalizeAggregation(quantilesTimingState(0.5)(x)) FROM (SELECT intDiv(number, 30000 AS d) AS k, number % d AS x FROM system.numbers LIMIT 100000) GROUP BY k WITH TOTALS ORDER BY k;