Emit SET NAMES for all MySQL connections w charset

The MySQL dialects will emit "SET NAMES" at the start of a connection when
charset is given to the MySQL driver, to appease an apparent behavior
observed in MySQL 8.0 that raises a collation error when a UNION includes
string columns unioned against columns of the form CAST(NULL AS CHAR(..)),
which is what SQLAlchemy's polymorphic_union function does.   The issue
seems to have affected PyMySQL for at least a year, however has recently
appeared as of mysqlclient 1.4.4 based on changes in how this DBAPI creates
a connection.  As the presence of this directive impacts three separate
MySQL charset settings which each have intricate effects based on their
presense,  SQLAlchemy will now emit the directive on new connections to
ensure correct behavior.

Fixes: #4804
Change-Id: If9d7ee00d0ccaf773972b564fe455e8e9edf6627

2 files changed, 17 insertions, 1 deletions

diff --git a/lib/sqlalchemy/dialects/mysql/mysqldb.py b/lib/sqlalchemy/dialects/mysql/mysqldb.py
index 744909170..1d4dba3c4 100644
--- a/lib/sqlalchemy/dialects/mysql/mysqldb.py
+++ b/lib/sqlalchemy/dialects/mysql/mysqldb.py
@@ -117,6 +117,22 @@ class MySQLDialect_mysqldb(MySQLDialect):
     def dbapi(cls):
         return __import__("MySQLdb")
 
+    def on_connect(self):
+        super_ = super(MySQLDialect_mysqldb, self).on_connect()
+
+        def on_connect(conn):
+            if super_ is not None:
+                super_(conn)
+
+            charset_name = conn.character_set_name()
+
+            if charset_name is not None:
+                cursor = conn.cursor()
+                cursor.execute("SET NAMES %s" % charset_name)
+                cursor.close()
+
+        return on_connect
+
     def do_ping(self, dbapi_connection):
         try:
             dbapi_connection.ping(False)
diff --git a/lib/sqlalchemy/testing/suite/test_types.py b/lib/sqlalchemy/testing/suite/test_types.py
index 3320dd93c..82a842102 100644
--- a/lib/sqlalchemy/testing/suite/test_types.py
+++ b/lib/sqlalchemy/testing/suite/test_types.py
@@ -733,7 +733,7 @@ class JSONTest(_LiteralRoundTripFixture, fixtures.TablesTest):
 
     def test_round_trip_custom_json(self):
         data_table = self.tables.data_table
-        data_element = self.data1
+        data_element = {"key1": "data1"}
 
         js = mock.Mock(side_effect=json.dumps)
         jd = mock.Mock(side_effect=json.loads)