ClickHouse/dbms/include/DB/DataStreams/FinalizingAggregatedBlockInputStream.h

63 lines
1.4 KiB
C
Raw Normal View History

2011-09-19 03:34:23 +00:00
#pragma once
#include <DB/Columns/ColumnAggregateFunction.h>
#include <DB/DataStreams/IProfilingBlockInputStream.h>
namespace DB
{
using Poco::SharedPtr;
/** Преобразует агрегатные функции (с промежуточным состоянием) в потоке блоков в конечные значения.
*/
class FinalizingAggregatedBlockInputStream : public IProfilingBlockInputStream
{
public:
FinalizingAggregatedBlockInputStream(BlockInputStreamPtr input_)
: input(input_)
{
children.push_back(input);
}
2012-10-20 02:10:47 +00:00
String getName() const { return "FinalizingAggregatedBlockInputStream"; }
BlockInputStreamPtr clone() { return new FinalizingAggregatedBlockInputStream(input); }
protected:
2011-09-19 03:34:23 +00:00
Block readImpl()
{
Block res = input->read();
if (!res)
return res;
size_t rows = res.rows();
size_t columns = res.columns();
for (size_t i = 0; i < columns; ++i)
{
ColumnWithNameAndType & column = res.getByPosition(i);
if (ColumnAggregateFunction * col = dynamic_cast<ColumnAggregateFunction *>(&*column.column))
{
ColumnAggregateFunction::Container_t & data = col->getData();
column.type = data[0]->getReturnType();
ColumnPtr finalized_column = column.type->createColumn();
for (size_t j = 0; j < rows; ++j)
finalized_column->insert(data[j]->getResult());
2012-10-20 02:10:47 +00:00
2011-09-19 03:34:23 +00:00
column.column = finalized_column;
}
}
return res;
}
private:
BlockInputStreamPtr input;
};
}