Added Rollup and Cube transforms.

dbms/src/Processors/Transforms/CubeTransform.cpp

+#include <Processors/Transforms/CubeTransform.h>
+
+namespace DB
+{
+
+static Chunk finalizeChunk(Chunk chunk)
+{
+    auto num_rows = chunk.getNumRows();
+    auto columns = chunk.detachColumns();
+
+    for (auto & column : columns)
+        if (auto * agg_function = typeid_cast<const ColumnAggregateFunction *>(column.get()))
+            column = agg_function->convertToValues();
+
+    return Chunk(std::move(columns), num_rows);
+}
+
+CubeTransform::CubeTransform(Block header, AggregatingTransformParamsPtr params_)
+    : IInflatingTransform(std::move(header), params_->getHeader())
+    , params(std::move(params_))
+    , keys(params->params.keys)
+{
+    if (keys.size() >= 8 * sizeof(mask))
+        throw Exception("Too many keys are used for CubeTransform.", ErrorCodes::LOGICAL_ERROR);
+}
+
+void CubeTransform::consume(Chunk chunk)
+{
+    consumed_chunk = std::move(chunk);
+    auto num_rows = current_chunk.getNumRows();
+    mask = (UInt64(1) << keys.size()) - 1;
+
+    current_columns = consumed_chunk.getColumns();
+    current_zero_columns.clear();
+    current_zero_columns.reserve(current_columns.size());
+
+    for (auto & column : current_columns)
+        current_zero_columns.emplace_back(column->cloneEmpty()->cloneResized(num_rows));
+}
+
+bool CubeTransform::canGenerate()
+{
+    return consumed_chunk;
+}
+
+Chunk CubeTransform::generate()
+{
+    auto gen_chunk = std::move(consumed_chunk);
+
+    if (mask)
+    {
+        --mask;
+
+        auto columns = current_columns;
+        for (size_t i = 0; i < columns.size(); ++i)
+            if (mask & (UInt64(1) << i))
+                columns[i] = current_zero_columns[i];
+
+        BlocksList cube_blocks = { getInputPort().getHeader().cloneWithColumns(columns) };
+        auto cube_block = params->aggregator.mergeBlocks(cube_blocks, false);
+
+        auto num_rows = cube_block.rows();
+        consumed_chunk = Chunk(cube_block.getColumns(), num_rows);
+    }
+
+    return finalizeChunk(std::move(gen_chunk));
+}
+
+}

dbms/src/Processors/Transforms/CubeTransform.h

+#pragma once
+#include <Processors/IInflatingTransform.h>
+#include <Processors/Transforms/AggregatingTransform.h>
+
+
+namespace DB
+{
+
+/// Takes blocks after grouping, with non-finalized aggregate functions.
+/// Calculates all subsets of columns and aggregates over them.
+class CubeTransform : public IInflatingTransform
+{
+public:
+    CubeTransform(Block header, AggregatingTransformParamsPtr params);
+
+protected:
+    void consume(Chunk chunk) override;
+
+    bool canGenerate() override;
+
+    Chunk generate() override;
+
+private:
+    AggregatingTransformParamsPtr params;
+    ColumnNumbers keys;
+
+    Chunk consumed_chunk;
+    Columns current_columns;
+    Columns current_zero_columns;
+
+    UInt64 mask = 0;
+};
+
+}

dbms/src/Processors/Transforms/RollupTransform.cpp

-//
-// Created by nik-kochetov on 3/18/19.
-//
+#include <Processors/Transforms/RollupTransform.h>
 
-#include "RollupTransform.h"
+namespace DB
+{
+
+static Chunk finalizeChunk(Chunk chunk)
+{
+    auto num_rows = chunk.getNumRows();
+    auto columns = chunk.detachColumns();
+
+    for (auto & column : columns)
+        if (auto * agg_function = typeid_cast<const ColumnAggregateFunction *>(column.get()))
+            column = agg_function->convertToValues();
+
+    return Chunk(std::move(columns), num_rows);
+}
+
+RollupTransform::RollupTransform(Block header, AggregatingTransformParamsPtr params_)
+    : IInflatingTransform(std::move(header)
+    , params_->getHeader())
+    , params(std::move(params_))
+    , keys(params->params.keys)
+{
+}
+
+void RollupTransform::consume(Chunk chunk)
+{
+    consumed_chunk = std::move(chunk);
+    last_removed_key = consumed_chunk.getNumColumns();
+}
+
+bool RollupTransform::canGenerate()
+{
+    return consumed_chunk;
+}
+
+Chunk RollupTransform::generate()
+{
+    auto gen_chunk = std::move(consumed_chunk);
+
+    if (last_removed_key)
+    {
+        --last_removed_key;
+
+        auto num_rows = gen_chunk.getNumRows();
+        auto columns = gen_chunk.getColumns();
+        columns[last_removed_key] = columns[last_removed_key]->cloneEmpty()->cloneResized(num_rows);
+
+        BlocksList rollup_blocks = { getInputPort().getHeader().cloneWithColumns(columns) };
+        auto rollup_block = params->aggregator.mergeBlocks(rollup_blocks, false);
+
+        num_rows = rollup_block.rows();
+        consumed_chunk = Chunk(rollup_block.getColumns(), num_rows);
+    }
+
+    return finalizeChunk(std::move(gen_chunk));
+}
+
+}

dbms/src/Processors/Transforms/RollupTransform.h

-//
-// Created by nik-kochetov on 3/18/19.
-//
-
-#ifndef CLICKHOUSE_ROLLUPTRANSFORM_H
-#define CLICKHOUSE_ROLLUPTRANSFORM_H
+#pragma once
+#include <Processors/IInflatingTransform.h>
+#include <Processors/Transforms/AggregatingTransform.h>
 
+namespace DB
+{
 
-class RollupTransform
+/// Takes blocks after grouping, with non-finalized aggregate functions.
+/// Calculates subtotals and grand totals values for a set of columns.
+class RollupTransform : public IInflatingTransform
 {
+public:
+    RollupTransform(Block header, AggregatingTransformParamsPtr params);
 
-};
+protected:
+    void consume(Chunk chunk) override;
+    bool canGenerate() override;
+    Chunk generate() override;
 
+private:
+    AggregatingTransformParamsPtr params;
+    ColumnNumbers keys;
+    Chunk consumed_chunk;
+    size_t last_removed_key = 0;
+};
 
-#endif //CLICKHOUSE_ROLLUPTRANSFORM_H
+}