Add timezone support, #170 (#171)

Author: bakwc

README.md

@@ -243,6 +243,8 @@ types_mapping:          # optional
 
 ignore_deletes: false    # optional, set to true to ignore DELETE operations
 
+mysql_timezone: 'UTC'    # optional, timezone for MySQL timestamp conversion (default: 'UTC')
+
 ```
 
 #### Required settings
@@ -267,6 +269,7 @@ ignore_deletes: false    # optional, set to true to ignore DELETE operations
 - `http_host`, `http_port` - http endpoint to control replication, use `/docs` for abailable commands
 - `types_mappings` - custom types mapping, eg. you can map char(36) to UUID instead of String, etc.
 - `ignore_deletes` - when set to `true`, DELETE operations in MySQL will be ignored during replication. This creates an append-only model where data is only added, never removed. In this mode, the replicator doesn't create a temporary database and instead replicates directly to the target database.
+- `mysql_timezone` - timezone to use for MySQL timestamp conversion to ClickHouse DateTime64. Default is `'UTC'`. Accepts any valid timezone name (e.g., `'America/New_York'`, `'Europe/London'`, `'Asia/Tokyo'`). This setting ensures proper timezone handling when converting MySQL timestamp fields to ClickHouse DateTime64 with timezone information.
 
 Few more tables / dbs examples:
 

mysql_ch_replicator/binlog_replicator.py

@@ -367,6 +367,7 @@ def __init__(self, settings: Settings):
             resume_stream=True,
             log_pos=log_pos,
             log_file=log_file,
+            mysql_timezone=settings.mysql_timezone,
         )
         self.last_state_update = 0
         self.last_binlog_clear_time = 0

mysql_ch_replicator/config.py

@@ -1,5 +1,6 @@
 import yaml
 import fnmatch
+import zoneinfo
 
 from dataclasses import dataclass
 
@@ -129,6 +130,7 @@ def __init__(self):
         self.target_databases = {}
         self.initial_replication_threads = 0
         self.ignore_deletes = False
+        self.mysql_timezone = 'UTC'
 
     def load(self, settings_file):
         data = open(settings_file, 'r').read()
@@ -155,6 +157,7 @@ def load(self, settings_file):
         self.target_databases = data.pop('target_databases', {})
         self.initial_replication_threads = data.pop('initial_replication_threads', 0)
         self.ignore_deletes = data.pop('ignore_deletes', False)
+        self.mysql_timezone = data.pop('mysql_timezone', 'UTC')
 
         indexes = data.pop('indexes', [])
         for index in indexes:
@@ -204,6 +207,16 @@ def validate_log_level(self):
         if self.log_level == 'debug':
             self.debug_log_level = True
 
+    def validate_mysql_timezone(self):
+        if not isinstance(self.mysql_timezone, str):
+            raise ValueError(f'mysql_timezone should be string and not {stype(self.mysql_timezone)}')
+        
+        # Validate timezone by attempting to import and check if it's valid
+        try:
+            zoneinfo.ZoneInfo(self.mysql_timezone)
+        except zoneinfo.ZoneInfoNotFoundError:
+            raise ValueError(f'invalid timezone: {self.mysql_timezone}. Use IANA timezone names like "UTC", "Europe/London", "America/New_York", etc.')
+
     def get_indexes(self, db_name, table_name):
         results = []
         for index in self.indexes:
@@ -235,3 +248,4 @@ def validate(self):
             raise ValueError(f'initial_replication_threads should be an integer, not {type(self.initial_replication_threads)}')
         if self.initial_replication_threads < 0:
             raise ValueError(f'initial_replication_threads should be non-negative')
+        self.validate_mysql_timezone()

mysql_ch_replicator/converter.py

@@ -214,7 +214,7 @@ def strip_sql_comments(sql_statement):
     return sqlparse.format(sql_statement, strip_comments=True).strip()
 
 
-def convert_timestamp_to_datetime64(input_str):
+def convert_timestamp_to_datetime64(input_str, timezone='UTC'):
 
     # Define the regex pattern
     pattern = r'^timestamp(?:\((\d+)\))?$'
@@ -226,9 +226,17 @@ def convert_timestamp_to_datetime64(input_str):
         # If a precision is provided, include it in the replacement
         precision = match.group(1)
         if precision is not None:
-            return f'DateTime64({precision})'
+            # Only add timezone info if it's not UTC (to preserve original behavior)
+            if timezone == 'UTC':
+                return f'DateTime64({precision})'
+            else:
+                return f'DateTime64({precision}, \'{timezone}\')'
         else:
-            return 'DateTime64'
+            # Only add timezone info if it's not UTC (to preserve original behavior)
+            if timezone == 'UTC':
+                return 'DateTime64'
+            else:
+                return f'DateTime64(3, \'{timezone}\')'
     else:
         raise ValueError(f"Invalid input string format: '{input_str}'")
 
@@ -372,7 +380,10 @@ def convert_type(self, mysql_type, parameters):
         if 'real' in mysql_type:
             return 'Float64'
         if mysql_type.startswith('timestamp'):
-            return convert_timestamp_to_datetime64(mysql_type)
+            timezone = 'UTC'
+            if self.db_replicator is not None:
+                timezone = self.db_replicator.config.mysql_timezone
+            return convert_timestamp_to_datetime64(mysql_type, timezone)
         if mysql_type.startswith('time'):
             return 'String'
         if 'varbinary' in mysql_type:

mysql_ch_replicator/pymysqlreplication/binlogstream.py

@@ -188,6 +188,7 @@ def __init__(
         ignore_decode_errors=False,
         verify_checksum=False,
         enable_logging=True,
+        mysql_timezone="UTC",
     ):
         """
         Attributes:
@@ -230,6 +231,7 @@ def __init__(
             verify_checksum: If true, verify events read from the binary log by examining checksums.
             enable_logging: When set to True, logs various details helpful for debugging and monitoring
                             When set to False, logging is disabled to enhance performance.
+            mysql_timezone: Timezone to use for MySQL timestamp conversion (e.g., 'UTC', 'America/New_York')
         """
 
         self.__connection_settings = connection_settings
@@ -254,6 +256,7 @@ def __init__(
         self.__ignore_decode_errors = ignore_decode_errors
         self.__verify_checksum = verify_checksum
         self.__optional_meta_data = False
+        self.__mysql_timezone = mysql_timezone
 
         # We can't filter on packet level TABLE_MAP and rotate event because
         # we need them for handling other operations
@@ -636,6 +639,7 @@ def fetchone(self):
                 self.__ignore_decode_errors,
                 self.__verify_checksum,
                 self.__optional_meta_data,
+                self.__mysql_timezone,
             )
 
             if binlog_event.event_type == ROTATE_EVENT:

mysql_ch_replicator/pymysqlreplication/event.py

@@ -28,6 +28,7 @@ def __init__(
         ignore_decode_errors=False,
         verify_checksum=False,
         optional_meta_data=False,
+        mysql_timezone="UTC",
     ):
         self.packet = from_packet
         self.table_map = table_map
@@ -39,6 +40,7 @@ def __init__(
         self._ignore_decode_errors = ignore_decode_errors
         self._verify_checksum = verify_checksum
         self._is_event_valid = None
+        self.mysql_timezone = mysql_timezone
         # The event have been fully processed, if processed is false
         # the event will be skipped
         self._processed = True

mysql_ch_replicator/pymysqlreplication/packet.py

@@ -75,6 +75,7 @@ def __init__(
         ignore_decode_errors,
         verify_checksum,
         optional_meta_data,
+        mysql_timezone="UTC",
     ):
         # -1 because we ignore the ok byte
         self.read_bytes = 0
@@ -128,6 +129,7 @@ def __init__(
             ignore_decode_errors=ignore_decode_errors,
             verify_checksum=verify_checksum,
             optional_meta_data=optional_meta_data,
+            mysql_timezone=mysql_timezone,
         )
         if not self.event._processed:
             self.event = None

mysql_ch_replicator/pymysqlreplication/row_event.py

@@ -1,6 +1,7 @@
 import struct
 import decimal
 import datetime
+import zoneinfo
 
 from pymysql.charset import charset_by_name
 from enum import Enum
@@ -100,6 +101,31 @@ def _is_null(null_bitmap, position):
             bit = ord(bit)
         return bit & (1 << (position % 8))
 
+    def _convert_timestamp_with_timezone(self, timestamp_value):
+        """
+        Convert timestamp from UTC to configured timezone
+
+        :param timestamp_value: Unix timestamp value
+        :return: datetime object in configured timezone
+        """
+        # Create UTC datetime first
+        utc_dt = datetime.datetime.utcfromtimestamp(timestamp_value)
+        
+        # If timezone is UTC, return timezone-aware UTC datetime
+        if self.mysql_timezone == "UTC":
+            return utc_dt.replace(tzinfo=datetime.timezone.utc)
+        
+        # Convert to configured timezone but keep timezone-aware
+        try:
+            # Start with UTC timezone-aware datetime
+            utc_dt_aware = utc_dt.replace(tzinfo=datetime.timezone.utc)
+            # Convert to target timezone
+            target_tz = zoneinfo.ZoneInfo(self.mysql_timezone)
+            return utc_dt_aware.astimezone(target_tz)
+        except zoneinfo.ZoneInfoNotFoundError:
+            # If timezone is invalid, fall back to UTC
+            return utc_dt.replace(tzinfo=datetime.timezone.utc)
+
     def _read_column_data(self, cols_bitmap, row_image_type=None):
         """Use for WRITE, UPDATE and DELETE events.
         Return an array of column data

test_mysql_ch_replicator.py

@@ -2804,3 +2804,111 @@ def test_json2():
     assert json.loads(ch.select(TEST_TABLE_NAME, "name='Peter'")[0]['data'])['в'] == 'б'
     db_replicator_runner.stop()
     binlog_replicator_runner.stop()
+
+def test_timezone_conversion():
+    """
+    Test that MySQL timestamp fields are converted to ClickHouse DateTime64 with custom timezone.
+    This test reproduces the issue from GitHub issue #170.
+    """
+    # Create a temporary config file with custom timezone
+    config_content = """
+mysql:
+  host: 'localhost'
+  port: 9306
+  user: 'root'
+  password: 'admin'
+
+clickhouse:
+  host: 'localhost'
+  port: 9123
+  user: 'default'
+  password: 'admin'
+
+binlog_replicator:
+  data_dir: '/app/binlog/'
+  records_per_file: 100000
+
+databases: '*test*'
+log_level: 'debug'
+mysql_timezone: 'America/New_York'
+"""
+    
+    # Create temporary config file
+    with tempfile.NamedTemporaryFile(mode='w', suffix='.yaml', delete=False) as f:
+        f.write(config_content)
+        temp_config_file = f.name
+    
+    try:
+        cfg = config.Settings()
+        cfg.load(temp_config_file)
+        
+        # Verify timezone is loaded correctly
+        assert cfg.mysql_timezone == 'America/New_York'
+        
+        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)
+
+        # Create table with timestamp fields
+        mysql.execute(f'''
+        CREATE TABLE `{TEST_TABLE_NAME}` (
+            id int NOT NULL AUTO_INCREMENT,
+            name varchar(255),
+            created_at timestamp NULL,
+            updated_at timestamp(3) NULL,
+            PRIMARY KEY (id)
+        );
+        ''')
+
+        # Insert test data with specific timestamp
+        mysql.execute(
+            f"INSERT INTO `{TEST_TABLE_NAME}` (name, created_at, updated_at) "
+            f"VALUES ('test_timezone', '2023-08-15 14:30:00', '2023-08-15 14:30:00.123');",
+            commit=True,
+        )
+
+        # Run replication
+        run_all_runner = RunAllRunner(cfg_file=temp_config_file)
+        run_all_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)
+
+        # Get the table structure from ClickHouse
+        table_info = ch.query(f'DESCRIBE `{TEST_TABLE_NAME}`')
+        
+        # Check that timestamp fields are converted to DateTime64 with timezone
+        created_at_type = None
+        updated_at_type = None
+        for row in table_info.result_rows:
+            if row[0] == 'created_at':
+                created_at_type = row[1]
+            elif row[0] == 'updated_at':
+                updated_at_type = row[1]
+        
+        # Verify the types include the timezone
+        assert created_at_type is not None
+        assert updated_at_type is not None
+        assert 'America/New_York' in created_at_type
+        assert 'America/New_York' in updated_at_type
+        
+        # Verify data was inserted correctly
+        results = ch.select(TEST_TABLE_NAME)
+        assert len(results) == 1
+        assert results[0]['name'] == 'test_timezone'
+        
+        run_all_runner.stop()
+        
+    finally:
+        # Clean up temporary config file
+        os.unlink(temp_config_file)