fix(mysql): handle the zero timestamp value

ibis/backends/mysql/__init__.py

@@ -7,11 +7,12 @@
 from typing import Iterable, Literal
 
 import sqlalchemy as sa
+from sqlalchemy.dialects import mysql
 
 import ibis.expr.datatypes as dt
 from ibis.backends.base.sql.alchemy import BaseAlchemyBackend
 from ibis.backends.mysql.compiler import MySQLCompiler
-from ibis.backends.mysql.datatypes import _type_from_cursor_info
+from ibis.backends.mysql.datatypes import MySQLDateTime, _type_from_cursor_info
 
 
 class Backend(BaseAlchemyBackend):
@@ -111,6 +112,16 @@ def connect(dbapi_connection, connection_record):
 
         super().do_connect(engine)
 
+    def _new_sa_metadata(self):
+        meta = super()._new_sa_metadata()
+
+        @sa.event.listens_for(meta, "column_reflect")
+        def column_reflect(inspector, table, column_info):
+            if isinstance(column_info["type"], mysql.DATETIME):
+                column_info["type"] = MySQLDateTime()
+
+        return meta
+
     def _metadata(self, query: str) -> Iterable[tuple[str, dt.DataType]]:
         if (
             re.search(r"^\s*SELECT\s", query, flags=re.MULTILINE | re.IGNORECASE)

ibis/backends/mysql/datatypes.py

@@ -7,6 +7,7 @@
 from sqlalchemy.dialects.mysql.base import MySQLDialect
 
 import ibis.expr.datatypes as dt
+from ibis.backends.base.sql.alchemy import to_sqla_type
 
 # binary character set
 # used to distinguish blob binary vs blob text
@@ -253,3 +254,20 @@ def mysql_tinyint(satype, nullable=True):
 @dt.dtype.register(mysql.BLOB)
 def mysql_blob(satype, nullable=True):
     return dt.Binary(nullable=nullable)
+
+
+class MySQLDateTime(mysql.DATETIME):
+    """Custom DATETIME type for MySQL that handles zero values."""
+
+    def result_processor(self, *_):
+        return lambda v: None if v == "0000-00-00 00:00:00" else v
+
+
+@dt.dtype.register(MySQLDateTime)
+def mysql_timestamp(_, nullable=True):
+    return dt.Timestamp(nullable=nullable)
+
+
+@to_sqla_type.register(MySQLDialect, dt.Timestamp)
+def _mysql_timestamp(*_):
+    return MySQLDateTime()