Support blob / binary data types

tap_mysql/discover_utils.py

@@ -53,7 +53,7 @@
 
 DATETIME_TYPES = {'datetime', 'timestamp', 'time', 'date'}
 
-BINARY_TYPES = {'binary', 'varbinary'}
+BINARY_TYPES = {'binary', 'varbinary', 'tinyblob', 'blob', 'mediumblob', 'longblob'}
 
 SPATIAL_TYPES = {'geometry', 'point', 'linestring',
                  'polygon', 'multipoint', 'multilinestring',

tests/integration/test_tap_mysql.py

@@ -1,3 +1,4 @@
+import codecs
 import os
 import unittest
 from unittest.mock import patch
@@ -335,6 +336,16 @@ def test_geometrycollection(self):
                           'sql-datatype': 'geometrycollection',
                           'datatype': 'geometrycollection'})
 
+    def test_blob(self):
+        actual = self.schema.properties['c_blob']
+        self.assertEqual(actual,
+                         Schema(['null', 'string'],
+                                format='binary',
+                                inclusion='available'))
+        self.assertEqual(self.get_metadata_for_column('c_blob'),
+                         {'selected-by-default': True,
+                          'sql-datatype': 'blob',
+                          'datatype': 'blob'})
 
 class TestSelectsAppropriateColumns(unittest.TestCase):
 
@@ -715,10 +726,10 @@ def setUp(self):
                 cursor.execute('CREATE TABLE binlog_1 (id int, updated datetime, '
                                'created_date Date)')
                 cursor.execute("""
-                    CREATE TABLE binlog_2 (id int, 
-                    updated datetime, 
-                    is_good bool default False, 
-                    ctime time, 
+                    CREATE TABLE binlog_2 (id int,
+                    updated datetime,
+                    is_good bool default False,
+                    ctime time,
                     cjson json)
                 """)
                 cursor.execute(
@@ -1012,7 +1023,7 @@ def test_binlog_stream_with_gtid(self):
         global SINGER_MESSAGES
 
         engine = os.getenv('TAP_MYSQL_ENGINE', MYSQL_ENGINE)
-        gtid = binlog.fetch_current_gtid_pos(self.conn, os.environ['TAP_MYSQL_ENGINE'])
+        gtid = binlog.fetch_current_gtid_pos(self.conn, engine)
 
         config = test_utils.get_db_config()
         config['use_gtid'] = True
@@ -1473,3 +1484,34 @@ def tearDown(self) -> None:
         with connect_with_backoff(self.conn) as open_conn:
             with open_conn.cursor() as cursor:
                 cursor.execute('DROP TABLE bit_booleans_table;')
+
+class TestBinaryMapping(unittest.TestCase):
+
+    def setUp(self):
+        self.conn = test_utils.get_test_connection()
+        self.blob_data = b'hello world'
+        with connect_with_backoff(self.conn) as conn:
+            with conn.cursor() as cursor:
+                cursor.execute("CREATE TABLE binary_table(id int, c_blob blob)")
+                cursor.execute("INSERT INTO binary_table VALUES (%s, %s)", (1, self.blob_data))
+        self.catalog = test_utils.discover_catalog(self.conn, {})
+
+    def tearDown(self):
+        with connect_with_backoff(self.conn) as open_conn:
+            with open_conn.cursor() as cursor:
+                cursor.execute('DROP TABLE binary_table')
+
+    def test_sync_messages_are_correct(self):
+        self.catalog.streams[0] = test_utils.set_replication_method_and_key(self.catalog.streams[0], 'FULL_TABLE', None)
+        self.catalog.streams[0] = test_utils.set_selected(self.catalog.streams[0], True)
+
+        global SINGER_MESSAGES
+        SINGER_MESSAGES.clear()
+
+        tap_mysql.do_sync(self.conn, {}, self.catalog, {})
+
+        record_messages = list(filter(lambda m: isinstance(m, singer.RecordMessage), SINGER_MESSAGES))
+
+        self.assertEqual(len(record_messages), 1)
+        self.assertEqual([rec.record for rec in record_messages],
+                         [{'id': 1, 'c_blob': codecs.encode(self.blob_data, 'hex').decode('utf-8').upper()}])