It seems MaterializeMySQL doesn't support Decimal data type when doing replication.
{
    if (!sync_quit && background_thread_pool)
    {
        sync_quit = true;
        background_thread_pool->join();
        client.disconnect();
    }
}
void MaterializeMySQLSyncThread::startSynchronization()
{
    const auto & mysql_server_version = checkVariableAndGetVersion(pool.get());
    background_thread_pool = std::make_unique<ThreadFromGlobalPool>(
        [this, mysql_server_version = mysql_server_version]() { synchronization(mysql_server_version); });
}
static inline void cleanOutdatedTables(const String & database_name, const Context & context)
{
    auto ddl_guard = DatabaseCatalog::instance().getDDLGuard(database_name, "");
    const DatabasePtr & clean_database = DatabaseCatalog::instance().getDatabase(database_name);
    for (auto iterator = clean_database->getTablesIterator(context); iterator->isValid(); iterator->next())
    {
        Context query_context = createQueryContext(context);
        String comment = "Materialize MySQL step 1: execute MySQL DDL for dump data";
        String table_name = backQuoteIfNeed(database_name) + "." + backQuoteIfNeed(iterator->name());
        tryToExecuteQuery(" DROP TABLE " + table_name, query_context, database_name, comment);
    }
}
static inline BlockOutputStreamPtr getTableOutput(const String & database_name, const String & table_name, Context & query_context, bool insert_materialized = false)