Method executeRefreshUnlocked

src/Storages/MaterializedView/RefreshTask.cpp:1003–1161

}

std::optional<UUID> RefreshTask::executeRefreshUnlocked(int32_t root_znode_version, std::vector<StorageID> deps, const String & log_comment, String & out_error_message)
{
    StorageID view_storage_id = view->getStorageID();
    LOG_DEBUG(getLogger(), "Refreshing view");
    execution.progress.reset();

    static constexpr bool internal = true;

    ContextMutablePtr refresh_context = view->getContext();
    ProcessList::EntryPtr process_list_entry;
    std::optional<StorageID> table_to_drop;
    auto new_table_id = StorageID::createEmpty();

    std::optional<QueryLogElement> query_log_elem;
    boost::intrusive_ptr<ASTInsertQuery> refresh_query;
    String query_for_logging;
    UInt64 normalized_query_hash = 0;
    std::shared_ptr<OpenTelemetry::SpanHolder> query_span = std::make_shared<OpenTelemetry::SpanHolder>("query");
    Stopwatch stopwatch;

    try
    {
        refresh_context = view->createRefreshContext(log_comment);

        syncDependenciesForRefresh(deps, refresh_context);

        if (!refresh_append)
        {
            refresh_context->setParentTable(view_storage_id.uuid);
            refresh_context->setDDLQueryCancellation(execution.cancel_ddl_queries.get_token());
            if (root_znode_version != -1)
                refresh_context->setDDLAdditionalChecksOnEnqueue({zkutil::makeCheckRequest(coordination.path, root_znode_version)});
        }

        {
            /// Create a table.
            query_for_logging = "(create target table)";
            normalized_query_hash = normalizedQueryHash(query_for_logging, false);
            QueryScope query_scope;
            std::tie(refresh_query, query_scope) = view->prepareRefresh(refresh_append, refresh_context, table_to_drop);
            new_table_id = refresh_query->table_id;

            /// Add the query to system.processes and allow it to be killed with KILL QUERY.
            query_for_logging = refresh_query->formatForLogging(
                refresh_context->getSettingsRef()[Setting::log_queries_cut_to_length]);
            normalized_query_hash = normalizedQueryHash(query_for_logging, false);

            process_list_entry = refresh_context->getProcessList().insert(
                query_for_logging, normalized_query_hash, refresh_query.get(), refresh_context, Stopwatch{CLOCK_MONOTONIC}.getStart(), internal);

            refresh_context->setProcessListElement(process_list_entry->getQueryStatus());
            refresh_context->setProgressCallback([this](const Progress & prog)
            {
                execution.progress.incrementPiecewiseAtomically(prog);
            });

            /// Run the query.