Support Interval Type

Author: rad-pat

Pipfile.lock

@@ -48,6 +48,8 @@ def escape_item(self, item):
             return self.escape_number(item)
         elif isinstance(item, decimal.Decimal):
             return self.escape_number(item)
+        elif isinstance(item, timedelta):
+             return self.escape_string(f'{item.total_seconds()} seconds') + '::interval'
         elif isinstance(item, (datetime, date, time, timedelta)):
             return self.escape_string(item.strftime("%Y-%m-%d %H:%M:%S"))
         else:

databend_sqlalchemy/connector.py

@@ -50,7 +50,9 @@
 )
 from sqlalchemy.engine import ExecutionContext, default
 from sqlalchemy.exc import DBAPIError, NoSuchTableError
+
 from .dml import Merge
+from .types import INTERVAL
 
 RESERVED_WORDS = {
     'Error', 'EOI', 'Whitespace', 'Comment', 'CommentBlock', 'Ident', 'ColumnPosition', 'LiteralString',
@@ -232,44 +234,8 @@ def process(value):
         return process
 
 
-class DatabendInterval(sqltypes.Interval):
-    """Stores interval as a datetime relative to epoch, see base implementation."""
-
-    _reg = re.compile(r"(\d+)-(\d+)-(\d+) (\d+):(\d+):(\d+)")
-
-    def result_processor(self, dialect, coltype):
-        def process(value):
-            if value is None:
-                return None
-            if isinstance(value, str):
-                m = self._reg.match(value)
-                if not m:
-                    raise ValueError(
-                        "could not parse %r as a datetime value" % (value,)
-                    )
-                groups = m.groups()
-                dt = datetime.datetime(*[
-                    int(groups[0] or self.epoch.year),
-                    int(groups[1] or self.epoch.month),
-                    int(groups[2] or self.epoch.day),
-                    int(groups[3] or 0),
-                    int(groups[4] or 0),
-                    int(groups[5] or 0),
-                ])
-            else:
-                dt = value
-            return dt - self.epoch
-
-        return process
-
-    def literal_processor(self, dialect):
-        def process(value):
-            if value is not None:
-                d = self.epoch + value
-                interval_str = d.isoformat(" ", timespec="microseconds")
-                return f"'{interval_str}'"
-
-        return process
+class DatabendInterval(INTERVAL):
+    render_bind_cast = True
 
 
 # Type converters
@@ -526,6 +492,9 @@ def visit_JSON(self, type_, **kw):
     def visit_TIME(self, type_, **kw):
         return "DATETIME"
 
+    def visit_INTERVAL(self, type, **kw):
+        return "INTERVAL"
+
 
 class DatabendDDLCompiler(compiler.DDLCompiler):
 

databend_sqlalchemy/databend_dialect.py

@@ -1,7 +1,7 @@
 
 from sqlalchemy.testing.provision import create_db
 from sqlalchemy.testing.provision import drop_db
-from sqlalchemy.testing.provision import configure_follower
+from sqlalchemy.testing.provision import configure_follower, update_db_opts
 
 
 @create_db.for_db("databend")
@@ -36,3 +36,8 @@ def _databend_drop_db(cfg, eng, ident):
 def _databend_configure_follower(config, ident):
     config.test_schema = "%s_test_schema" % ident
     config.test_schema_2 = "%s_test_schema_2" % ident
+
+# Uncomment to debug SQL Statements in tests
+# @update_db_opts.for_db("databend")
+# def _mssql_update_db_opts(db_url, db_opts):
+#     db_opts["echo"] = True

databend_sqlalchemy/provision.py

@@ -0,0 +1,109 @@
+
+from __future__ import annotations
+
+import datetime as dt
+import re
+from typing import Optional, Type, Any
+
+from sqlalchemy.engine.interfaces import Dialect
+from sqlalchemy.sql import sqltypes
+from sqlalchemy.sql import type_api
+
+INTERVAL_RE = re.compile(
+    r"^"
+    r"(?:(?P<days>-?\d+) (days? ?))?"
+    r"(?:(?P<sign>[-+])?"
+    r"(?P<hours>\d+):"
+    r"(?P<minutes>\d\d):"
+    r"(?P<seconds>\d\d)"
+    r"(?:\.(?P<microseconds>\d{1,6}))?"
+    r")?$"
+)
+
+
+# ToDo - This is perhaps how numeric should be defined
+# class NUMERIC(sqltypes.Numeric):
+#     def result_processor(self, dialect, type_):
+#
+#         orig = super().result_processor(dialect, type_)
+#
+#         def process(value):
+#             if value is not None:
+#                 if self.decimal_return_scale:
+#                     value = decimal.Decimal(f'{value:.{self.decimal_return_scale}f}')
+#                 else:
+#                     value = decimal.Decimal(value)
+#             if orig:
+#                 return orig(value)
+#             return value
+#
+#         return process
+
+
+class INTERVAL(type_api.NativeForEmulated, sqltypes._AbstractInterval):
+    """Databend INTERVAL type."""
+
+    __visit_name__ = "INTERVAL"
+    native = True
+
+    def __init__(
+        self, precision: Optional[int] = None, fields: Optional[str] = None
+    ) -> None:
+        """Construct an INTERVAL.
+
+        :param precision: optional integer precision value
+        :param fields: string fields specifier.  allows storage of fields
+         to be limited, such as ``"YEAR"``, ``"MONTH"``, ``"DAY TO HOUR"``,
+         etc.
+
+        """
+        self.precision = precision
+        self.fields = fields
+
+    @classmethod
+    def adapt_emulated_to_native(
+        cls, interval: sqltypes.Interval, **kw: Any  # type: ignore[override]
+    ) -> INTERVAL:
+        return INTERVAL(precision=interval.second_precision)
+
+    @property
+    def _type_affinity(self) -> Type[sqltypes.Interval]:
+        return sqltypes.Interval
+
+    def as_generic(self, allow_nulltype: bool = False) -> sqltypes.Interval:
+        return sqltypes.Interval(native=True, second_precision=self.precision)
+
+    @property
+    def python_type(self) -> Type[dt.timedelta]:
+        return dt.timedelta
+
+    def literal_processor(
+        self, dialect: Dialect
+    ) -> Optional[type_api._LiteralProcessorType[dt.timedelta]]:
+        def process(value: dt.timedelta) -> str:
+            return f"to_interval('{value.total_seconds()} seconds')"
+
+        return process
+
+    # ToDo - If BendSQL returns a timedelta for interval, then we should be able to remove this method
+    def result_processor(
+        self, dialect: Dialect, coltype: Any
+    ) -> type_api._ResultProcessorType[dt.timedelta]:
+
+        def process(value: Any) -> Optional[dt.timedelta]:
+            """Parse a duration string and return a datetime.timedelta."""
+            if value is None:
+                return None
+
+            match = INTERVAL_RE.match(value)
+            if match:
+                kw = match.groupdict()
+                sign = -1 if kw.pop("sign", "+") == "-" else 1
+                if kw.get("microseconds"):
+                    kw["microseconds"] = kw["microseconds"].ljust(6, "0")
+                kw = {k: float(v.replace(",", ".")) for k, v in kw.items() if v is not None}
+                days = dt.timedelta(kw.pop("days", 0.0) or 0.0)
+
+                return days + sign * dt.timedelta(**kw)
+
+        return process

databend_sqlalchemy/types.py

@@ -16,13 +16,10 @@
 from sqlalchemy.testing.suite import JoinTest as _JoinTest
 from sqlalchemy.testing.suite import BizarroCharacterFKResolutionTest as _BizarroCharacterFKResolutionTest
 from sqlalchemy.testing.suite import ServerSideCursorsTest as _ServerSideCursorsTest
-from sqlalchemy.testing.suite import IntervalTest as _IntervalTest
-from sqlalchemy.testing.suite import PrecisionIntervalTest as _PrecisionIntervalTest
 from sqlalchemy.testing.suite import EnumTest as _EnumTest
 from sqlalchemy import types as sql_types
 from sqlalchemy import testing, select
 from sqlalchemy.testing import config, eq_
-from databend_sqlalchemy.databend_dialect import DatabendInterval
 
 
 class ComponentReflectionTest(_ComponentReflectionTest):
@@ -283,47 +280,6 @@ def test_roundtrip_fetchmany(self):
         pass
 
 
-class IntervalTest(_IntervalTest):
-    __backend__ = True
-    datatype = DatabendInterval
-
-    @testing.skip("databend")  # Skipped because cannot figure out the literal() part
-    def test_arithmetic_operation_literal_interval(self, connection):
-        pass
-
-    @testing.skip("databend")  # Skipped because cannot figure out the literal() part
-    def test_arithmetic_operation_table_interval_and_literal_interval(
-        self, connection, arithmetic_table_fixture
-    ):
-        pass
-
-    @testing.skip("databend")  # Skipped because cannot figure out the literal() part
-    def test_arithmetic_operation_table_date_and_literal_interval(
-        self, connection, arithmetic_table_fixture
-    ):
-        pass
-
-
-class PrecisionIntervalTest(_PrecisionIntervalTest):
-    __backend__ = True
-    datatype = DatabendInterval
-
-    @testing.skip("databend")  # Skipped because cannot figure out the literal() part
-    def test_arithmetic_operation_literal_interval(self, connection):
-        pass
-
-    @testing.skip("databend")  # Skipped because cannot figure out the literal() part
-    def test_arithmetic_operation_table_interval_and_literal_interval(
-        self, connection, arithmetic_table_fixture
-    ):
-        pass
-
-    @testing.skip("databend")  # Skipped because cannot figure out the literal() part
-    def test_arithmetic_operation_table_date_and_literal_interval(
-        self, connection, arithmetic_table_fixture
-    ):
-        pass
-
 class EnumTest(_EnumTest):
     __backend__ = True