Method insertAndRun

src/Server/AsyncQueryManager.cpp:40–85

}

void AsyncQueryManager::insertAndRun(
    String & query,
    ASTPtr ast,
    ContextMutablePtr ctx,
    ReadBuffer * istr,
    SendAsyncQueryIdCallback send_async_query_id,
    AsyncQueryHandlerFunc && func)
{
    if (pool)
    {
        String id = UUIDHelpers::UUIDToString(UUIDHelpers::generateV4());
        ctx->setAsyncQueryId(id);
        AsyncQueryStatus status;
        status.set_id(id);
        status.set_query_id(ctx->getClientInfo().current_query_id);
        status.set_status(AsyncQueryStatus::NotStarted);
        auto c_time = time(nullptr);
        status.set_start_time(c_time);
        status.set_update_time(c_time);
        status.set_max_execution_time(ctx->getSettingsRef().max_execution_time.totalSeconds());
        ctx->getCnchCatalog()->setAsyncQueryStatus(id, status);

        send_async_query_id(id);

        pool->scheduleOrThrowOnError(make_copyable_function<void()>([query = std::move(query),
                                                                     ast = std::move(ast),
                                                                     context = std::move(ctx),
                                                                     istr,
                                                                     func = std::move(func),
                                                                     id = std::move(id),
                                                                     status = std::move(status)]() mutable {
            setThreadName("async_query");
            status.set_status(AsyncQueryStatus::Running);
            status.set_update_time(time(nullptr));
            context->getCnchCatalog()->setAsyncQueryStatus(id, status);

            std::optional<CurrentThread::QueryScope> query_scope;
            query_scope.emplace(context);
            func(query, ast, context, istr);
        }));
    }
    else
    {
        func(query, ast, ctx, istr);
    }
}

} // namespace DB