Method doResyncTable

src/Databases/MySQL/MaterializedMySQLSyncThreadManager.cpp:1459–1551

}

void MaterializedMySQLSyncThreadManager::doResyncTable(const String & table_name)
{
    String tmp_table_suffix = "_CHTMP";
    String query_prefix = "EXTERNAL DDL FROM MySQL(" + backQuoteIfNeed(storage_id.database_name) + ", "
                                    + backQuoteIfNeed(materialized_mysql_ptr->getMysqlDatabaseInfo().mysql_database_name) + ") ";
    Stopwatch timer;
    LOG_DEBUG(log, "Start to resync table {}", table_name);
    if (auto materialize_mysql_log = getContext()->getCloudMaterializedMySQLLog())
    {
        auto current_log = MaterializedMySQL::createMaterializeMySQLLog(MaterializedMySQLLogElement::SYNC_MANAGER, MaterializedMySQLLogElement::RESYNC_TABLE, storage_id.database_name);
        current_log.resync_table = table_name;
        materialize_mysql_log->add(current_log);
    }

    /// Check table existence
    auto connection = mysql_component->pool.get();
    if (!checkTableExistence(connection, materialized_mysql_ptr->getMysqlDatabaseInfo().mysql_database_name, table_name, getContext()->getSettingsRef()))
        throw Exception(ErrorCodes::UNKNOWN_TABLE, "Resync table {} doesn't exist in mysql.", table_name);

    String full_mysql_table_name = backQuoteIfNeed(materialized_mysql_ptr->getMysqlDatabaseInfo().mysql_database_name) + "." + backQuoteIfNeed(table_name);
    String tmp_table_name = table_name + tmp_table_suffix;
    String full_mysql_tmp_table_name = backQuoteIfNeed(materialized_mysql_ptr->getMysqlDatabaseInfo().mysql_database_name) + "." + backQuoteIfNeed(tmp_table_name);
    Block show_create_table_header{
        {std::make_shared<DataTypeString>(), "Table"},
        {std::make_shared<DataTypeString>(), "Create Table"},
    };

    StreamSettings mysql_input_stream_settings(getContext()->getSettingsRef(), false, true);
    MySQLBlockInputStream show_create_table(
        connection, "SHOW CREATE TABLE " + full_mysql_table_name, show_create_table_header, mysql_input_stream_settings);

    Block create_query_block = show_create_table.read();

    if (!create_query_block || create_query_block.rows() != 1)
        throw Exception("LOGICAL ERROR mysql show create return more rows.", ErrorCodes::LOGICAL_ERROR);

    String create_query = create_query_block.getByName("Create Table").column->getDataAt(0).toString();
    LOG_DEBUG(log, "The original create table query of resync table {} is {}", table_name, create_query);
    create_query = create_query.replace(create_query.find(table_name), table_name.length(), tmp_table_name);
    LOG_DEBUG(log, "The temp create table query of resync temp table {} is {}", tmp_table_name, create_query);

    auto query_context = MaterializedMySQL::createQueryContext(getContext());
    {
        String comment = "Materialize MySQL resync table step 1: execute MySQL DDL for dump data to tmp table";
        tryToExecuteQueryWithTxn(query_prefix + " DROP TABLE IF EXISTS " + full_mysql_tmp_table_name, query_context, storage_id.database_name, comment);
        tryToExecuteQueryWithTxn(query_prefix + " " + create_query, query_context, storage_id.database_name, comment);

        auto query_context_inter = MaterializedMySQL::createQueryContext(query_context);
        auto txn = createTransactionForMySQL(query_context);
        SCOPE_EXIT({
            if (txn)
                query_context->getCnchTransactionCoordinator().finishTransaction(txn);
        });

        query_context_inter->setCurrentTransaction(txn);

        auto out = std::make_shared<CountingBlockOutputStream>(getTableOutput(storage_id.database_name, tmp_table_name, query_context_inter));
        StreamSettings inner_mysql_input_stream_settings(getContext()->getSettingsRef());