Method processOrdinaryQuery

src/Server/TCPHandler.cpp:1481–1566

void TCPHandler::processOrdinaryQuery(QueryState & state)
{
    auto & pipeline = state.io.pipeline;

    /// Send header-block, to allow client to prepare output format for data to send.
    {
        const auto & header = pipeline.getHeader();

        if (!header.empty())
        {
            sendData(state, header);
        }
    }

    {
        PullingAsyncPipelineExecutor executor(pipeline);
        pipeline.setConcurrencyControl(state.query_context->getSettingsRef()[Setting::use_concurrency_control]);
        CurrentMetrics::Increment query_thread_metric_increment{CurrentMetrics::QueryThread};

        try
        {
            Block block;
            while (executor.pull(block, interactive_delay / 1000))
            {
                bool stop_read_return_partial_result = false;
                {
                    std::lock_guard lock(*callback_mutex);
                    receivePacketsExpectCancel(state);
                    stop_read_return_partial_result = state.stop_read_return_partial_result;
                }

                if (stop_read_return_partial_result)
                {
                    executor.cancelReading();
                }

                {
                    std::lock_guard lock(*callback_mutex);

                    if (after_send_progress.elapsed() / 1000 >= interactive_delay)
                    {
                        /// Some time passed and there is a progress.
                        after_send_progress.restart();
                        sendProgress(state);
                        sendSelectProfileEvents(state);
                    }

                    sendLogs(state);

                    // Block might be empty in case of timeout, i.e. there is no data to process
                    if (!block.empty() && !state.io.null_format)
                        sendData(state, block);
                }
            }
        }
        catch (...)
        {
            executor.cancel();