Method executeDDLAtomic

src/Databases/MySQL/MaterializeMySQLSyncThread.cpp:667–725

}

void MaterializeMySQLSyncThread::executeDDLAtomic(const QueryEvent & query_event)
{
    try
    {
        String query = query_event.query;
        auto ddl_params = parseMySQLDDLQuery(query_event.query);
        if (ddl_params.query_type == MySQLDDLQueryParams::UNKNOWN_TYPE)
        {
            LOG_DEBUG(log, "Skip unsupported MySQL DDL query: {}", query_event.query);
            return;
        }
        else if (!ddl_params.execute_table.empty())
        {
            auto ddl_database_name =  ddl_params.execute_database.empty() ? query_event.schema: ddl_params.execute_database;
            if (ddl_database_name != mysql_database_name
                || (assigned_materialized_table != ddl_params.execute_table && ddl_params.query_type != MySQLDDLQueryParams::CREATE_TABLE))
            {
                LOG_DEBUG(log, "Skip MySQL DDL: {}", query_event.query);
                return;
            }
        }

        auto database = DatabaseCatalog::instance().getDatabase(database_name, getContext());
        auto materialized_mysql = dynamic_cast<DatabaseCloudMaterializedMySQL*>(database.get());
        if (!materialized_mysql)
            throw Exception("Database should be CloudMaterializedMySQL", ErrorCodes::LOGICAL_ERROR);
        auto server_host_port = materialized_mysql->getServerClientOfManager();
        auto server_client = getContext()->getCnchServerClient(server_host_port.getHost(), server_host_port.rpc_port);
        if (!server_client)
            throw Exception("Failed to get server client for " + server_host_port.toDebugString(), ErrorCodes::LOGICAL_ERROR);

        auto pos = database_name.find("_" + thread_key);
        if (pos == std::string::npos)
            throw Exception(
                "Local MaterializedMySQL database #" + database_name + " has no thread key suffix #" + thread_key,
                 ErrorCodes::LOGICAL_ERROR
            );
        String cnch_database_name = database_name.substr(0, pos);

        auto position = client.getPosition();
        MySQLBinLogInfo binlog;
        binlog.binlog_file = position.binlog_name;
        binlog.binlog_position = position.binlog_pos;
        binlog.executed_gtid_set = position.gtid_sets.toString();

        server_client->submitMaterializedMySQLDDLQuery(cnch_database_name, thread_key, query, binlog);
    }
    catch (Exception & exception)
    {
        exception.addMessage("While executing MYSQL_QUERY_EVENT. The query: " + query_event.query);

        tryLogCurrentException(log);

        /// If some DDL query was not successfully parsed and executed
        /// Then replication may fail on next binlog events anyway
        if (exception.code() != ErrorCodes::SYNTAX_ERROR)
            throw;
    }