Method init

src/DataStreams/MergingSortedBlockInputStream.cpp:31–80

}

void MergingSortedBlockInputStream::init(MutableColumns & merged_columns)
{
    /// Read the first blocks, initialize the queue.
    if (first)
    {
        first = false;

        for (size_t i = 0; i < source_blocks.size(); ++i)
        {
            Block & block = source_blocks[i];

            if (block)
                continue;

            block = children[i]->read();

            const size_t rows = block.rows();

            if (rows == 0)
                continue;

            if (expected_block_size < rows)
                expected_block_size = std::min(rows, max_block_size);

            cursors[i] = SortCursorImpl(block, description, i);
            has_collation |= cursors[i].has_collation;
        }

        if (has_collation)
            queue_with_collation = SortingHeap<SortCursorWithCollation>(cursors);
        else
            queue_without_collation = SortingHeap<SortCursor>(cursors);
    }

    /// Let's check that all source blocks have the same structure.
    for (const auto & block : source_blocks)
    {
        if (!block)
            continue;

        assertBlocksHaveEqualStructure(block, header, getName());
    }

    merged_columns.resize(num_columns);
    for (size_t i = 0; i < num_columns; ++i)
    {
        merged_columns[i] = header.safeGetByPosition(i).column->cloneEmpty();
        merged_columns[i]->reserve(expected_block_size);
    }
}


Block MergingSortedBlockInputStream::readImpl()