Method executeByInsertOverwrite

src/Storages/StorageMaterializedView.cpp:654–749

}

void StorageMaterializedView::executeByInsertOverwrite(AsyncRefreshParamPtr param, ContextMutablePtr local_context)
{
    if (param->insert_overwrite_query.empty())
        throw Exception("materialized view refresh insert overwrite query is empty.", ErrorCodes::LOGICAL_ERROR);

    auto create_command_context = [local_context](std::string sub_id) {
        auto command_context = Context::createCopy(local_context);
        command_context->setCurrentTransaction(nullptr, false);
        command_context->setCurrentVW(nullptr);
        command_context->setCurrentWorkerGroup(nullptr);
        command_context->makeSessionContext();
        command_context->makeQueryContext();
        String query_id = fmt::format("{}_{}", command_context->getCurrentQueryId(), sub_id);
        command_context->setCurrentQueryId(query_id);
        auto settings = local_context->getSettings();
        command_context->setSettings(settings);
        command_context->setSetting("enable_materialized_view_rewrite", false);
        return command_context;
    };

    auto insert_overwrite_context = create_command_context("mv_insert_overwrite");
    auto & txn_coordinator = local_context->getCnchTransactionCoordinator();
    auto server_txn = txn_coordinator.createTransaction(CreateTransactionOption().setType(CnchTransactionType::Implicit));
    const_cast<Context &>(*insert_overwrite_context).setCurrentTransaction(server_txn);

    /// Add commit and abort function for mv meta
    auto mv_commit_func = [param = param, this](ContextPtr context) {
        return context->getCnchCatalog()->constructMvMetaRequests(
            UUIDHelpers::UUIDToString(this->getStorageUUID()), param->part_diff->add_partitions, param->part_diff->drop_partitions, toString(context->getTimestamp()));
    };
    auto mv_abort_func = [param = param, this](ContextPtr context) {
        return context->getCnchCatalog()->constructMvMetaRequests(
            UUIDHelpers::UUIDToString(this->getStorageUUID()), param->part_diff->add_partitions, param->part_diff->drop_partitions, toString(context->getTimestamp()));
    };
    insert_overwrite_context->getCurrentTransaction()->addCommitAbortFunc(mv_commit_func, mv_abort_func);

    /// Insert refresh task log
    auto start_time = std::chrono::system_clock::now();
    insertRefreshTaskLog(param, RefreshViewTaskStatus::START, true, start_time, local_context);

    std::optional<Exception> exception;
    ThreadFromGlobalPool async_thread([&]() {
        try
        {
            std::optional<CurrentThread::QueryScope> query_scope;
            query_scope.emplace(insert_overwrite_context);
            CurrentThread::get().pushTenantId(insert_overwrite_context->getSettingsRef().tenant_id);

            LOG_DEBUG(log, "refresh sync materialized view refresh insert overwite query: {}", param->insert_overwrite_query);
            BlockIO insert_io;
            try
            {
                insert_io = executeQuery(param->insert_overwrite_query, insert_overwrite_context, false);
                if (insert_io.pipeline.initialized())
                {
                    auto & pipeline = insert_io.pipeline;
                    PullingAsyncPipelineExecutor executor(pipeline);
                    Block block;