It seems MaterializeMySQL doesn't support Decimal data type when doing replication.
{
    bool opened_transaction = false;
    mysqlxx::PoolWithFailover::Entry connection;
    while (!isCancelled())
    {
        try
        {
            connection = pool.get();
            opened_transaction = false;
            MaterializeMetadata metadata(
                connection, getDatabase(database_name).getMetadataPath() + "/.metadata",
                mysql_database_name, opened_transaction, mysql_version);
            if (!metadata.need_dumping_tables.empty())
            {
                Position position;
                position.update(metadata.binlog_position, metadata.binlog_file, metadata.executed_gtid_set);
                metadata.transaction(position, [&]()
                {
                    cleanOutdatedTables(database_name, global_context);
                    dumpDataForTables(connection, metadata, query_prefix, database_name, mysql_database_name, global_context, [this] { return isCancelled(); });
                });
                const auto & position_message = [&]()
                {
                    std::stringstream ss;