Method processInsertQuery

src/Server/TCPHandler.cpp:1374–1478

void TCPHandler::processInsertQuery(QueryState & state)
{
    size_t num_threads = state.io.pipeline.getNumThreads();

    auto run_executor = [&](auto & executor, Block processed_data)
    {
        try
        {
            /// Made above the rest of the lines,
            /// so that in case of `start` function throws an exception,
            /// client receive exception before sending data.
            executor.start();

            if (!processed_data.empty())
                executor.push(std::move(processed_data));
            else
                startInsertQuery(state);

            while (receivePacketsExpectDataConcurrentWithExecutor(state))
            {
                executor.push(std::move(state.block_for_insert));

                std::lock_guard lock(*callback_mutex);
                sendLogs(state);
                sendInsertProfileEvents(state);
            }

            executor.finish();
        }
        catch (...)
        {
            executor.cancel();
            throw;
        }
    };

    Block processed_block;
    const auto & settings = state.query_context->getSettingsRef();

    auto * insert_queue = state.query_context->tryGetAsynchronousInsertQueue();
    const auto & insert_query = assert_cast<const ASTInsertQuery &>(*state.parsed_query);

    bool async_insert_enabled = settings[Setting::async_insert];
    if (insert_query.table_id)
        if (auto table = DatabaseCatalog::instance().tryGetTable(insert_query.table_id, state.query_context))
            async_insert_enabled |= table->areAsynchronousInsertsEnabled();

    if (insert_queue && async_insert_enabled && !insert_query.select)
    {
        auto result = processAsyncInsertQuery(state, *insert_queue);
        if (result.status == AsynchronousInsertQueue::PushResult::OK)
        {
            /// Reset pipeline because it may hold write lock for some storages.
            state.io.resetPipeline(/*cancel=*/true);
            if (settings[Setting::wait_for_async_insert])
            {
                size_t timeout_ms = settings[Setting::wait_for_async_insert_timeout].totalMilliseconds();
                auto wait_status = result.future.wait_for(std::chrono::milliseconds(timeout_ms));