Merge branch 'master' into optimizer_auto_call

Author: bakwc

README.md

@@ -3,7 +3,7 @@
 [![Release][release-image]][releases]
 [![License][license-image]][license]
 
-[release-image]: https://img.shields.io/badge/release-0.0.32-blue.svg?style=flat
+[release-image]: https://img.shields.io/badge/release-0.0.35-blue.svg?style=flat
 [releases]: https://github.com/bakwc/mysql_ch_replicator/releases
 
 [license-image]: https://img.shields.io/badge/license-MIT-blue.svg?style=flat

mysql_ch_replicator/converter.py

@@ -1,3 +1,4 @@
+import struct
 import json
 import sqlparse
 import re
@@ -27,6 +28,58 @@ def convert_bytes(obj):
         return obj
 
 
+def parse_mysql_point(binary):
+    """
+    Parses the binary representation of a MySQL POINT data type
+    and returns a tuple (x, y) representing the coordinates.
+
+    :param binary: The binary data representing the POINT.
+    :return: A tuple (x, y) with the coordinate values.
+    """
+    if binary is None:
+        return 0, 0
+
+    if len(binary) == 21:
+        # No SRID. Proceed as per WKB POINT
+        # Read the byte order
+        byte_order = binary[0]
+        if byte_order == 0:
+            endian = '>'
+        elif byte_order == 1:
+            endian = '<'
+        else:
+            raise ValueError("Invalid byte order in WKB POINT")
+        # Read the WKB Type
+        wkb_type = struct.unpack(endian + 'I', binary[1:5])[0]
+        if wkb_type != 1:  # WKB type 1 means POINT
+            raise ValueError("Not a WKB POINT type")
+        # Read X and Y coordinates
+        x = struct.unpack(endian + 'd', binary[5:13])[0]
+        y = struct.unpack(endian + 'd', binary[13:21])[0]
+    elif len(binary) == 25:
+        # With SRID included
+        # First 4 bytes are the SRID
+        srid = struct.unpack('>I', binary[0:4])[0]  # SRID is big-endian
+        # Next byte is byte order
+        byte_order = binary[4]
+        if byte_order == 0:
+            endian = '>'
+        elif byte_order == 1:
+            endian = '<'
+        else:
+            raise ValueError("Invalid byte order in WKB POINT")
+        # Read the WKB Type
+        wkb_type = struct.unpack(endian + 'I', binary[5:9])[0]
+        if wkb_type != 1:  # WKB type 1 means POINT
+            raise ValueError("Not a WKB POINT type")
+        # Read X and Y coordinates
+        x = struct.unpack(endian + 'd', binary[9:17])[0]
+        y = struct.unpack(endian + 'd', binary[17:25])[0]
+    else:
+        raise ValueError("Invalid binary length for WKB POINT")
+    return (x, y)
+
+
 def strip_sql_name(name):
     name = name.strip()
     if name.startswith('`'):
@@ -64,9 +117,11 @@ def __init__(self, db_replicator: 'DbReplicator' = None):
         self.db_replicator = db_replicator
 
     def convert_type(self, mysql_type, parameters):
-
         is_unsigned = 'unsigned' in parameters.lower()
 
+        if mysql_type == 'point':
+            return 'Tuple(x Float32, y Float32)'
+
         if mysql_type == 'int':
             if is_unsigned:
                 return 'UInt32'
@@ -89,6 +144,8 @@ def convert_type(self, mysql_type, parameters):
             return 'Date32'
         if mysql_type == 'tinyint(1)':
             return 'Bool'
+        if mysql_type == 'bit(1)':
+            return 'Bool'
         if mysql_type == 'bool':
             return 'Bool'
         if 'smallint' in mysql_type:
@@ -123,27 +180,31 @@ def convert_type(self, mysql_type, parameters):
             return 'Float32'
         if 'double' in mysql_type:
             return 'Float64'
-        if 'integer' in mysql_type or 'int(' in mysql_type:
-            if is_unsigned:
-                return 'UInt32'
-            return 'Int32'
         if 'bigint' in mysql_type:
             if is_unsigned:
                 return 'UInt64'
             return 'Int64'
+        if 'integer' in mysql_type or 'int(' in mysql_type:
+            if is_unsigned:
+                return 'UInt32'
+            return 'Int32'
         if 'real' in mysql_type:
             return 'Float64'
         if mysql_type.startswith('time'):
             return 'String'
         if 'varbinary' in mysql_type:
             return 'String'
+        if 'binary' in mysql_type:
+            return 'String'
         raise Exception(f'unknown mysql type "{mysql_type}"')
 
     def convert_field_type(self, mysql_type, mysql_parameters):
         mysql_type = mysql_type.lower()
         mysql_parameters = mysql_parameters.lower()
         not_null = 'not null' in mysql_parameters
         clickhouse_type = self.convert_type(mysql_type, mysql_parameters)
+        if 'Tuple' in clickhouse_type:
+            not_null = True
         if not not_null:
             clickhouse_type = f'Nullable({clickhouse_type})'
         return clickhouse_type
@@ -195,6 +256,9 @@ def convert_record(self, mysql_record, mysql_field_types, clickhouse_field_types
                 if 'UInt64' in clickhouse_field_type and clickhouse_field_value < 0:
                     clickhouse_field_value = 18446744073709551616 + clickhouse_field_value
 
+            if 'point' in mysql_field_type:
+                clickhouse_field_value = parse_mysql_point(clickhouse_field_value)
+
             clickhouse_record.append(clickhouse_field_value)
         return tuple(clickhouse_record)
 

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "mysql-ch-replicator"
-version = "0.0.32"
+version = "0.0.35"
 description = "Tool for replication of MySQL databases to ClickHouse"
 authors = ["Filipp Ozinov <[email protected]>"]
 license = "MIT"

test_mysql_ch_replicator.py

@@ -694,6 +694,72 @@ def test_numeric_types_and_limits():
     assert_wait(lambda: len(ch.select(TEST_TABLE_NAME, 'test7=18446744073709551586')) == 2)
 
 
+def test_different_types_2():
+    cfg = config.Settings()
+    cfg.load(CONFIG_FILE)
+
+    mysql = mysql_api.MySQLApi(
+        database=None,
+        mysql_settings=cfg.mysql,
+    )
+
+    ch = clickhouse_api.ClickhouseApi(
+        database=TEST_DB_NAME,
+        clickhouse_settings=cfg.clickhouse,
+    )
+
+    prepare_env(cfg, mysql, ch)
+
+    mysql.execute("SET sql_mode = 'ALLOW_INVALID_DATES';")
+
+    mysql.execute(f'''
+CREATE TABLE {TEST_TABLE_NAME} (
+    `id` int unsigned NOT NULL AUTO_INCREMENT,
+    test1 bit(1),
+    test2 point,
+    test3 binary(16),
+    PRIMARY KEY (id)
+); 
+    ''')
+
+    mysql.execute(
+        f"INSERT INTO {TEST_TABLE_NAME} (test1, test2, test3) VALUES "
+        f"(0, POINT(10.0, 20.0), 'azaza');",
+        commit=True,
+    )
+
+    binlog_replicator_runner = BinlogReplicatorRunner()
+    binlog_replicator_runner.run()
+    db_replicator_runner = DbReplicatorRunner(TEST_DB_NAME)
+    db_replicator_runner.run()
+
+    assert_wait(lambda: TEST_DB_NAME in ch.get_databases())
+
+    ch.execute_command(f'USE {TEST_DB_NAME}')
+
+    assert_wait(lambda: TEST_TABLE_NAME in ch.get_tables())
+    assert_wait(lambda: len(ch.select(TEST_TABLE_NAME)) == 1)
+
+    mysql.execute(
+        f"INSERT INTO {TEST_TABLE_NAME} (test1, test2) VALUES "
+        f"(1, POINT(15.0, 14.0));",
+        commit=True,
+    )
+    assert_wait(lambda: len(ch.select(TEST_TABLE_NAME)) == 2)
+    assert_wait(lambda: len(ch.select(TEST_TABLE_NAME, 'test1=True')) == 1)
+
+    assert ch.select(TEST_TABLE_NAME, 'test1=True')[0]['test2']['x'] == 15.0
+    assert ch.select(TEST_TABLE_NAME, 'test1=False')[0]['test2']['y'] == 20.0
+    assert ch.select(TEST_TABLE_NAME, 'test1=False')[0]['test3'] == 'azaza\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
+
+    mysql.execute(
+        f"INSERT INTO {TEST_TABLE_NAME} (test1, test2) VALUES "
+        f"(0, NULL);",
+        commit=True,
+    )
+    assert_wait(lambda: len(ch.select(TEST_TABLE_NAME)) == 3)
+
+
 def test_json():
     cfg = config.Settings()
     cfg.load(CONFIG_FILE)