It seems MaterializeMySQL doesn't support Decimal data type when doing replication.
{"binlog_format", "binlog_format='ROW'"},
            {"binlog_row_image", "binlog_row_image='FULL'"},
            {"default_authentication_plugin", "default_authentication_plugin='mysql_native_password'"}
        };
        ColumnPtr variable_name_column = variables_block.getByName("Variable_name").column;
        for (size_t index = 0; index < variables_block.rows(); ++index)
        {
            const auto & error_message_it = variables_error_message.find(variable_name_column->getDataAt(index).toString());
            if (error_message_it != variables_error_message.end())
                variables_error_message.erase(error_message_it);
        }
        bool first = true;
        std::stringstream error_message;
        error_message << "Illegal MySQL variables, the MaterializeMySQL engine requires ";
        for (const auto & [variable_name, variable_error_message] : variables_error_message)
        {
            error_message << (first ? "" : ", ") << variable_error_message;
            if (first)
                first = false;
        }
        throw Exception(error_message.str(), ErrorCodes::ILLEGAL_MYSQL_VARIABLE);
    }
    Block version_header{{std::make_shared<DataTypeString>(), "version"}};
    MySQLBlockInputStream version_input(connection, "SELECT version() AS version;", version_header, DEFAULT_BLOCK_SIZE);
    Block version_block = version_input.read();
    if (!version_block || version_block.rows() != 1)
        throw Exception("LOGICAL ERROR: cannot get mysql version.", ErrorCodes::LOGICAL_ERROR);
    return version_block.getByPosition(0).column->getDataAt(0).toString();
}
MaterializeMySQLSyncThread::MaterializeMySQLSyncThread(