Method runImpl

src/CloudServices/CnchRefreshMaterializedViewThread.cpp:130–185

}

void CnchRefreshMaterializedViewThread::runImpl()
{
    UInt64 next_delay_seconds = 10;

    try
    {
        runHeartbeatTask();

        auto istorage = getStorageFromCatalog();
        auto & storage = checkAndGetMaterializedViewTable(istorage);
        auto target_istorage = storage.getTargetTable();
        auto & target = checkAndGetCnchTable(target_istorage);
        auto storage_settings = target.getSettings();

        if (istorage->is_dropped)
        {
            LOG_DEBUG(log, "Table was dropped, wait for removing...");
            scheduled_task->scheduleAfter(10 * 1000);
            return;
        }

        {
            std::lock_guard lock(worker_pool_mutex);
            vw_name = storage_settings->cnch_vw_write;
            vm_handle = getContext()->getVirtualWarehousePool().get(vw_name);
        }

        bool success = false;

        try
        {
            int max_bg_task_num = storage_settings->max_refresh_materialized_view_task_num;
            if (running_tasks < max_bg_task_num)
            {
                success = startRefreshTask(istorage, storage);
            }
            else
            {
                LOG_DEBUG(log, "Too many cnch materialized view refresh tasks (current: {}, max: {})", running_tasks, max_bg_task_num);
            }
        }
        catch (...)
        {
            tryLogCurrentException(__PRETTY_FUNCTION__);
        }

        next_delay_seconds = checkAndGetRefreshInterval(storage);
        scheduled_task->scheduleAfter(next_delay_seconds * 1000);
    }
    catch (...)
    {
        tryLogCurrentException(__PRETTY_FUNCTION__);

        scheduled_task->scheduleAfter(next_delay_seconds * 1000);
    }
}

bool CnchRefreshMaterializedViewThread::constructAndScheduleRefreshTasks(StoragePtr & istorage, StorageMaterializedView & storage)