Method tryGenerate

src/Server/DistributedQuery/StreamingExchangeSource.cpp:237–338

}

std::optional<Chunk> StreamingExchangeSource::tryGenerate()
{
    if (!was_on_start_called)
    {
        was_on_start_called = true;
        onStart();
        return Chunk(); /// Empty chunk means we need to be called again
    }

    if (output_finished)
    {
        LOG_TRACE(log, "NoMoreDataNeeded from exchange stream {}, total rows: {}, bytes: {}", stream_name, rows_read, bytes_read);

        sendNoMoreDataNeeded();
        finished_reading = true;
        return {};
    }

    LOG_TEST(log, "Reading from exchange stream {}", stream_name);

    if (packet_receive_state == ReceivingHeader)
        tryReadHeader();

    if (packet_receive_state == ReceivingBody)
        tryReadBody();

    /// If a whole packet has been read, we can parse it.
    if (!packet_in)
        return Chunk(); /// Empty chunk means that we currently heve no data but we have not finished yet.

    UInt64 flags = 0;
    readVarUInt(flags, *packet_in);
    const bool final_chunk = (flags & 1);
    const bool has_aggregated_chunk_info = (flags & 2);
    UInt64 num_rows = 0;
    readVarUInt(num_rows, *packet_in);
    UInt64 num_columns = 0;
    readVarUInt(num_columns, *packet_in);
    UInt64 chunk_num = 0;
    if (has_aggregated_chunk_info)
        readVarUInt(chunk_num, *packet_in);

    /// The final packet is the empty end-of-stream marker. A final packet carrying rows would have
    /// them dropped once finished_reading is set, so reject it as a protocol violation.
    if (final_chunk && num_rows != 0)
        throw Exception(ErrorCodes::UNEXPECTED_PACKET_FROM_CLIENT,
            "Final data packet on exchange stream {} carries {} rows; it must be empty", stream_name, num_rows);

    /// A data packet must carry exactly the header's columns, or values would be dropped while the
    /// row count is kept. A header-less stream (e.g. SELECT count()) sends rows with zero columns.
    const size_t expected_columns = output.getHeader().columns();
    if (num_rows != 0 && num_columns != expected_columns)
        throw Exception(ErrorCodes::UNEXPECTED_PACKET_FROM_CLIENT,
            "Data packet on exchange stream {} carries {} rows with {} columns, but the stream header has {} columns",
            stream_name, num_rows, num_columns, expected_columns);

    std::optional<Chunk> result;
    if (num_columns != 0)