Method finish

src/QueryPipeline/RemoteQueryExecutor.cpp:772–847

}

void RemoteQueryExecutor::finish()
{
    LockAndBlocker guard(was_cancelled_mutex);

    /** If one of:
      * - nothing started to do;
      * - received all packets before EndOfStream;
      * - received exception from one replica;
      * - received an unknown packet from one replica;
      * then you do not need to read anything.
      */
    if (!isQueryPending() || hasThrownException() || was_cancelled)
        return;

    /// To make sure finish is only called once
    SCOPE_EXIT({ finished = true; });

    /** If you have not read all the data yet, but they are no longer needed.
      * This may be due to the fact that the data is sufficient (for example, when using LIMIT).
      */

    /// Send the request to abort the execution of the request, if not already sent.
    tryCancel("Cancelling query because enough data has been read");

    /// If connections weren't created yet, query wasn't sent or was already finished, nothing to do.
    if (!connections || !sent_query || finished)
        return;

    /// Get the remaining packets so that there is no out of sync in the connections to the replicas.
    /// We do this manually instead of calling drain() because we want to process Log, ProfileEvents and Progress
    /// packets that had been sent before the connection is fully finished in order to have final statistics of what
    /// was executed in the remote queries
    while (connections->hasActiveConnections() && !finished)
    {
        Packet packet = connections->receivePacket();

        switch (packet.type)
        {
            case Protocol::Server::EndOfStream:
                finished = true;
                break;

            case Protocol::Server::Exception:
                got_exception_from_replica = true;
                packet.exception->rethrow();
                break;

            case Protocol::Server::Log:
                /// Pass logs from remote server to client
                if (auto log_queue = CurrentThread::getInternalTextLogsQueue())
                    log_queue->pushBlock(std::move(packet.block));
                break;

            case Protocol::Server::ProfileEvents:
                /// Pass profile events from remote server to client
                if (auto profile_queue = CurrentThread::getInternalProfileEventsQueue())
                    if (!profile_queue->emplace(std::move(packet.block)))
                        throw Exception(ErrorCodes::SYSTEM_ERROR, "Could not push into profile queue");