Method iterate

src/Databases/MySQL/MaterializedMySQLSyncThreadManager.cpp:682–735

}

bool MaterializedMySQLSyncThreadManager::iterate()
{
    if (!initialized)
    {
        initialize();
        initialized = true;
    }

    /// When materialized_tables_list is empty, reinit
    if (materialized_tables_list.empty())
        initialized = false;

    /// Judge whether sync thread manager should stop according to settings & skip_sync_failed_tables/skipped_unsupported_tables
    {
        std::lock_guard lock(status_info_mutex);
        if ((!materialized_mysql_ptr->getMaterializeMySQLSettings()->skip_sync_failed_tables && !sync_failed_tables.empty())
            || (!materialized_mysql_ptr->getMaterializeMySQLSettings()->skip_unsupported_tables && !skipped_unsupported_tables.empty()))
        {
            stopSyncThreads();
            threads_info.clear();
            initialized = false;
            setSyncType(MaterializedMySQLSyncType::ExceptionStopSync);
            return true;
        }
    }
    /// Make a copy of `threads_info` for each iteration to avoid hold the lock for a long time
    /// while the DDL actions would change the `threads_info`
    std::vector<SyncThreadScheduleInfoPtr> current_threads_info;
    {
        std::lock_guard lock(status_info_mutex);
        for (auto & thread_info : threads_info)
        {
            std::lock_guard thread_lock(thread_info->mutex);
            if (shouldSyncTable(thread_info->assigned_materialized_table, false, false))
                current_threads_info.emplace_back(thread_info);
        }
    }

    setSyncType(MaterializedMySQLSyncType::Syncing);
    std::lock_guard resync_lock(resync_mutex);
    for (auto & thread_info : current_threads_info)
    {
        std::lock_guard thread_lock(thread_info->mutex);
        if (resync_tables.count(thread_info->assigned_materialized_table))
            continue;

        if (thread_info->worker_client)
            checkStatusOfSyncThread(thread_info);

        if (!thread_info->worker_client)
            scheduleSyncThreadToWorker(thread_info);
    }
    return false;
}

void MaterializedMySQLSyncThreadManager::initialize()
{