Method fillData

src/Storages/System/StorageSystemAsyncLoader.cpp:79–139

}

void StorageSystemAsyncLoader::fillData(MutableColumns & res_columns, ContextPtr context, const ActionsDAG::Node *, std::vector<UInt8>) const
{
    TimePoint now = std::chrono::system_clock::now();

    AsyncLoader & async_loader = context->getAsyncLoader();

    for (const auto & state : async_loader.getJobStates())
    {
        Array dependencies;
        dependencies.reserve(state.job->dependencies.size());
        for (const auto & dep : state.job->dependencies)
            dependencies.emplace_back(dep->jobId());

        TimePoint started = state.job->startTime();
        TimePoint finished = state.job->finishTime();
        TimePoint last = finished != TimePoint{} ? finished : now;
        TimeDuration elapsed = started != TimePoint{} ? last - started : TimeDuration{0};
        double elapsed_sec = static_cast<double>(std::chrono::duration_cast<std::chrono::nanoseconds>(elapsed).count()) * 1e-9;

        Field ready_seqno;
        if (state.ready_seqno)
            ready_seqno = state.ready_seqno;

        Field exception;
        if (state.job->exception())
        {
            try
            {
                std::rethrow_exception(state.job->exception());
            }
            catch (...)
            {
                exception = getCurrentExceptionMessage(false);
            }
        }

        size_t i = 0;
        res_columns[i++]->insert(state.job->name);
        res_columns[i++]->insert(state.job->jobId());
        res_columns[i++]->insert(dependencies);
        res_columns[i++]->insert(state.dependencies_left);
        res_columns[i++]->insert(static_cast<Int8>(state.job->status()));
        res_columns[i++]->insert(state.is_executing);
        res_columns[i++]->insert(state.is_blocked);
        res_columns[i++]->insert(state.is_ready);
        res_columns[i++]->insert(elapsed_sec);
        res_columns[i++]->insert(state.job->pool());
        res_columns[i++]->insert(async_loader.getPoolName(state.job->pool()));
        res_columns[i++]->insert(async_loader.getPoolPriority(state.job->pool()).value);
        res_columns[i++]->insert(state.job->executionPool());
        res_columns[i++]->insert(async_loader.getPoolName(state.job->executionPool()));
        res_columns[i++]->insert(async_loader.getPoolPriority(state.job->executionPool()).value);
        res_columns[i++]->insert(ready_seqno);
        res_columns[i++]->insert(state.job->waitersCount());
        res_columns[i++]->insert(exception);
        res_columns[i++]->insert(timeInMicroseconds(state.job->scheduleTime()));
        res_columns[i++]->insert(optionalTimeInMicroseconds(state.job->enqueueTime()));
        res_columns[i++]->insert(optionalTimeInMicroseconds(state.job->startTime()));