Method executeByDropInsert

src/Storages/StorageMaterializedView.cpp:498–652

}

void StorageMaterializedView::executeByDropInsert(AsyncRefreshParamPtr param, ContextMutablePtr local_context)
{
    if (param->drop_partition_query.empty() && param->insert_select_query.empty())
        throw Exception("Materialized view refresh drop partition annd insert select query is empty.", ErrorCodes::LOGICAL_ERROR);

    /// BEGIN
    auto & txn_coordinator = local_context->getCnchTransactionCoordinator();
    auto explicit_txn = txn_coordinator.createTransaction(CreateTransactionOption().setType(CnchTransactionType::Explicit));
    const_cast<Context &>(*local_context).setCurrentTransaction(explicit_txn, true);

    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 start_time = std::chrono::system_clock::now();
    insertRefreshTaskLog(param, RefreshViewTaskStatus::START, false, start_time, local_context);

    std::optional<Exception> exception;
    /// DROP PARTITION
    if (!param->drop_partition_query.empty())
    {
        LOG_DEBUG(log, "refresh sync materialized view execute drop partition query: {}", param->drop_partition_query);
        auto drop_context = create_command_context("mv_drop");
        ThreadFromGlobalPool drop_thread([&]() {
            std::optional<CurrentThread::QueryScope> query_scope;
            query_scope.emplace(drop_context);
            BlockIO drop_io;
            try
            {
                try
                {
                    drop_io = executeQuery(param->drop_partition_query, drop_context);
                    if (drop_io.pipeline.initialized())
                    {
                        auto & pipeline = drop_io.pipeline;
                        PullingAsyncPipelineExecutor executor(pipeline);
                        Block block;
                        while (executor.pull(block))
                        {
                        }
                    }
                    else if (drop_io.in)
                    {
                        AsynchronousBlockInputStream async_in(drop_io.in);
                        async_in.readPrefix();
                        while (true)
                        {
                            const auto block = async_in.read();