Function tryQueueQuery

src/Interpreters/executeQuery.cpp:284–330

}

void tryQueueQuery(ContextMutablePtr context, ASTPtr & query_ast)
{
    ASTType ast_type;
    try {
        ast_type = query_ast->getType();
    } catch (...) {
        LOG_DEBUG(&Poco::Logger::get("executeQuery"), "only queue dml query");
        return;
    }
    auto worker_group_handler = context->tryGetCurrentWorkerGroup();
    if (ast_type != ASTType::ASTSelectQuery && ast_type != ASTType::ASTSelectWithUnionQuery && ast_type != ASTType::ASTInsertQuery
        && ast_type != ASTType::ASTDeleteQuery && ast_type != ASTType::ASTUpdateQuery)
    {
        LOG_DEBUG(&Poco::Logger::get("executeQuery"), "only queue dml query");
        return;
    }
    if (worker_group_handler)
    {
        Stopwatch queue_watch;
        queue_watch.start();
        auto query_queue = context->getQueueManager();
        auto query_id = context->getCurrentQueryId();
        const auto & vw_name = worker_group_handler->getVWName();
        const auto & wg_name = worker_group_handler->getID();
        context->getWorkerStatusManager()->updateVWWorkerList(worker_group_handler->getHostWithPortsVec(), vw_name, wg_name);
        auto queue_info = std::make_shared<QueueInfo>(query_id, vw_name, wg_name, context);
        auto queue_result = query_queue->enqueue(queue_info, context->getSettingsRef().query_queue_timeout_ms);
        if (queue_result == QueueResultStatus::QueueSuccess)
        {
            auto current_vw = context->tryGetCurrentVW();
            if (current_vw)
            {
                context->setCurrentWorkerGroup(current_vw->getWorkerGroup(wg_name));
            }
            LOG_DEBUG(&Poco::Logger::get("executeQuery"), "query queue run time : {} ms", queue_watch.elapsedMilliseconds());
        }
        else
        {
            LOG_ERROR(&Poco::Logger::get("executeQuery"), "query queue result : {}", queueResultStatusToString(queue_result));
            throw Exception(
                ErrorCodes::CNCH_QUEUE_QUERY_FAILURE,
                "query queue failed for query_id {}: {}",
                query_id,
                queueResultStatusToString(queue_result));
        }
    }
}

bool needThrowRootCauseError(const Context * context, int & error_code, String & error_messge)
{