Method generate

src/Processors/Sources/SourceFromInputStream.cpp:134–198

}

Chunk SourceFromInputStream::generate()
{
    if (is_stream_finished)
        return {};

    if (!is_stream_started)
    {
        stream->readPrefix();
        is_stream_started = true;
    }

    auto block = stream->read();
    if (!block && !isCancelled())
    {
        if (rows_before_limit)
        {
            const auto & info = stream->getProfileInfo();
            if (info.hasAppliedLimit())
                rows_before_limit->add(info.getRowsBeforeLimit());
        }

        stream->readSuffix();

        if (auto totals_block = stream->getTotals())
        {
            if (totals_block.rows() > 0) /// Sometimes we can get empty totals. Skip it.
            {
                totals.setColumns(totals_block.getColumns(), totals_block.rows());
                has_totals = true;
            }
        }

        if (auto extremes_block = stream->getExtremes())
        {
            if (extremes_block.rows() > 0) /// Sometimes we can get empty extremes. Skip it.
            {
                extremes.setColumns(extremes_block.getColumns(), extremes_block.rows());
                has_extremes = true;
            }
        }

        is_stream_finished = true;
        return {};
    }

    if (isCancelled())
        return {};

#ifndef NDEBUG
    assertBlocksHaveEqualStructure(getPort().getHeader(), block, "SourceFromInputStream");
#endif

    UInt64 num_rows = block.rows();
    Chunk chunk(block.getColumns(), num_rows);

    if (force_add_aggregating_info || has_aggregate_functions)
    {
        auto info = std::make_shared<AggregatedChunkInfo>();