Method processOrdinaryQuery

src/Server/TCPHandler.cpp:787–859

void TCPHandler::processOrdinaryQuery()
{
    OpenTelemetrySpanHolder span(__PRETTY_FUNCTION__);

    /// Pull query execution result, if exists, and send it to network.
    if (state.io.in)
    {
        if (query_context->getSettingsRef().allow_experimental_query_deduplication)
            sendPartUUIDs();

        /// This allows the client to prepare output format
        if (Block header = state.io.in->getHeader())
            sendData(header);

        /// Use of async mode here enables reporting progress and monitoring client cancelling the query
        AsynchronousBlockInputStream async_in(state.io.in);

        async_in.readPrefix();
        while (true)
        {
            if (isQueryCancelled())
            {
                async_in.cancel(false);
                break;
            }

            if (after_send_progress.elapsed() / 1000 >= query_context->getSettingsRef().interactive_delay)
            {
                /// Some time passed.
                after_send_progress.restart();
                sendProgress();
            }

            sendLogs();

            if (async_in.poll(query_context->getSettingsRef().interactive_delay / 1000))
            {
                const auto block = async_in.read();
                if (!block)
                    break;

                if (!state.io.null_format)
                    sendData(block);
            }
        }
        async_in.readSuffix();

        /** When the data has run out, we send the profiling data and totals up to the terminating empty block,
          * so that this information can be used in the suffix output of stream.
          * If the request has been interrupted, then sendTotals and other methods should not be called,
          * because we have not read all the data.
          */
        if (!isQueryCancelled())
        {
            sendTotals(state.io.in->getTotals());
            sendExtremes(state.io.in->getExtremes());
            sendProfileInfo(state.io.in->getProfileInfo());
            sendProgress();