Function transformToSingleBlockSources

src/Storages/StorageExecutable.cpp:65–96

namespace
{
    void transformToSingleBlockSources(Pipes & inputs)
    {
        size_t inputs_size = inputs.size();
        for (size_t i = 0; i < inputs_size; ++i)
        {
            auto && input = inputs[i];
            QueryPipeline input_pipeline(std::move(input));
            PullingPipelineExecutor input_pipeline_executor(input_pipeline);

            auto header = input_pipeline_executor.getHeader();
            auto result_block = header.cloneEmpty();

            size_t result_block_columns = result_block.columns();

            Block result;
            while (input_pipeline_executor.pull(result))
            {
                for (size_t result_block_index = 0; result_block_index < result_block_columns; ++result_block_index)
                {
                    auto & block_column = result.safeGetByPosition(result_block_index);
                    auto & result_block_column = result_block.safeGetByPosition(result_block_index);

                    auto mutable_column = IColumn::mutate(std::move(result_block_column.column));
                    mutable_column->insertRangeFrom(*block_column.column, 0, block_column.column->size());
                    result_block_column.column = std::move(mutable_column);
                }
            }

            auto source = std::make_shared<SourceFromSingleChunk>(std::make_shared<const Block>(std::move(result_block)));
            inputs[i] = Pipe(std::move(source));
        }
    }
}

StorageExecutable::StorageExecutable(