2019-03-19 13:36:19 +00:00
|
|
|
#include <Processors/Transforms/RollupTransform.h>
|
2019-04-08 13:02:19 +00:00
|
|
|
#include <Processors/Transforms/TotalsHavingTransform.h>
|
2019-03-19 11:08:33 +00:00
|
|
|
|
2019-03-19 13:36:19 +00:00
|
|
|
namespace DB
|
|
|
|
{
|
|
|
|
|
|
|
|
RollupTransform::RollupTransform(Block header, AggregatingTransformParamsPtr params_)
|
2019-08-22 08:26:50 +00:00
|
|
|
: IAccumulatingTransform(std::move(header), params_->getHeader())
|
2019-03-19 13:36:19 +00:00
|
|
|
, params(std::move(params_))
|
|
|
|
, keys(params->params.keys)
|
|
|
|
{
|
|
|
|
}
|
|
|
|
|
|
|
|
void RollupTransform::consume(Chunk chunk)
|
|
|
|
{
|
2019-08-22 08:26:50 +00:00
|
|
|
consumed_chunks.emplace_back(std::move(chunk));
|
2019-03-19 13:36:19 +00:00
|
|
|
}
|
|
|
|
|
2019-08-22 08:26:50 +00:00
|
|
|
Chunk RollupTransform::merge(Chunks && chunks, bool final)
|
2019-03-19 13:36:19 +00:00
|
|
|
{
|
2019-08-22 08:26:50 +00:00
|
|
|
BlocksList rollup_blocks;
|
|
|
|
for (auto & chunk : chunks)
|
|
|
|
rollup_blocks.emplace_back(getInputPort().getHeader().cloneWithColumns(chunk.detachColumns()));
|
|
|
|
|
|
|
|
auto rollup_block = params->aggregator.mergeBlocks(rollup_blocks, final);
|
|
|
|
auto num_rows = rollup_block.rows();
|
|
|
|
return Chunk(rollup_block.getColumns(), num_rows);
|
2019-03-19 13:36:19 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
Chunk RollupTransform::generate()
|
|
|
|
{
|
2019-08-22 08:26:50 +00:00
|
|
|
if (!consumed_chunks.empty())
|
|
|
|
{
|
|
|
|
if (consumed_chunks.size() > 1)
|
|
|
|
rollup_chunk = merge(std::move(consumed_chunks), false);
|
|
|
|
else
|
|
|
|
rollup_chunk = std::move(consumed_chunks.front());
|
|
|
|
|
|
|
|
consumed_chunks.clear();
|
|
|
|
}
|
|
|
|
|
|
|
|
auto gen_chunk = std::move(rollup_chunk);
|
2019-03-19 13:36:19 +00:00
|
|
|
|
|
|
|
if (last_removed_key)
|
|
|
|
{
|
|
|
|
--last_removed_key;
|
2019-04-18 12:09:03 +00:00
|
|
|
auto key = keys[last_removed_key];
|
2019-03-19 13:36:19 +00:00
|
|
|
|
|
|
|
auto num_rows = gen_chunk.getNumRows();
|
|
|
|
auto columns = gen_chunk.getColumns();
|
2019-04-18 12:09:03 +00:00
|
|
|
columns[key] = columns[key]->cloneEmpty()->cloneResized(num_rows);
|
2019-03-19 13:36:19 +00:00
|
|
|
|
2019-08-22 08:29:29 +00:00
|
|
|
Chunks chunks;
|
|
|
|
chunks.emplace_back(std::move(columns), num_rows);
|
|
|
|
rollup_chunk = merge(std::move(chunks), false);
|
2019-03-19 13:36:19 +00:00
|
|
|
}
|
|
|
|
|
2019-04-08 13:02:19 +00:00
|
|
|
finalizeChunk(gen_chunk);
|
|
|
|
return gen_chunk;
|
2019-03-19 13:36:19 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
}
|