Method startTask

src/Server/StatelessWorker/StatelessTaskExecutor.cpp:40–147

}

StatelessTaskExecutor::Result StatelessTaskExecutor::startTask(const String & unique_task_id, const DistributedQueryTaskDescription & task_description, const String & unique_temp_file_path)
{
    /// `unique_task_id` is unique per task, so a repeated start (e.g. a coordinator retry) is the same
    /// task. Running it twice would double-write exchanges and temp files and orphan the original from
    /// cancel/forget, so treat a duplicate start as a no-op.
    {
        std::lock_guard lock(tasks_mutex);
        if (tasks.contains(unique_task_id))
        {
            LOG_WARNING(getLogger("StatelessTaskExecutor"), "Ignoring duplicate start for already running task {}", unique_task_id);
            return Result::Ok;
        }
    }

    ContextPtr global_context = Context::getGlobalContextInstance();
    ContextMutablePtr query_context = Context::createCopy(global_context);
    query_context->makeQueryContext();
    {
        ClientInfo client_info;
        client_info.current_query_id = unique_task_id;
        client_info.query_kind = ClientInfo::QueryKind::SECONDARY_QUERY;
        client_info.initial_query_id = task_description.initial_query_id;
        query_context->setClientInfo(client_info);
    }

    /// Apply the initiator's settings so the worker honors query limits and execution-affecting
    /// settings. Limits (e.g. max_rows_to_read, max_rows_in_join) are enforced per task, so each
    /// fragment stays under the limit but the whole query may exceed it by up to the bucket count.
    /// Force make_distributed_plan off: the worker runs an already-split local fragment.
    query_context->applySettingsChanges(task_description.settings_changes);
    query_context->setSetting("make_distributed_plan", false);

    auto [object_storage, object_storage_path] = getObjectStorageForTemporaryFiles(unique_temp_file_path, query_context);

    std::shared_ptr<std::promise<String>> task_promise = std::make_shared<std::promise<String>>();
    auto task_state = std::make_shared<TaskState>();
    task_state->completion_future = task_promise->get_future();

    {
        std::lock_guard lock(tasks_mutex);
        /// If two starts of the same id race, keep the first; overwriting would orphan the running task.
        if (!tasks.try_emplace(unique_task_id, task_state).second)
        {
            LOG_WARNING(getLogger("StatelessTaskExecutor"), "Ignoring duplicate start for already running task {}", unique_task_id);
            return Result::Ok;
        }
    }

    /// Callback for periodic cancellation check
    auto is_task_cancelled = [cancelled = task_state->cancelled]() -> bool
    {
        return *cancelled;
    };

    auto update_progress = [task_progress = task_state->progress](const Progress & progress)
    {
        task_progress->incrementPiecewiseAtomically(progress);
    };