Merge pull request #75 from datafold/normalize_types_refactor

Refactor Normalize-types into normalize_timestamp() normalize_number()

Author: erezsh

data_diff/database.py

@@ -131,6 +131,10 @@ def __post_init__(self):
 class UnknownColType(ColType):
     text: str
 
+    def __post_init__(self):
+        logger.warn(f"Column of type '{self.text}' has no compatibility handling. "
+                     "If encoding/formatting differs between databases, it may result in false positives.")
+
 
 class AbstractDatabase(ABC):
     @abstractmethod
@@ -173,16 +177,24 @@ def close(self):
         "Close connection(s) to the database instance. Querying will stop functioning."
         ...
 
+
     @abstractmethod
-    def normalize_value_by_type(value: str, coltype: ColType) -> str:
-        """Creates an SQL expression, that converts 'value' to a normalized representation.
+    def normalize_timestamp(self, value: str, coltype: ColType) -> str:
+        """Creates an SQL expression, that converts 'value' to a normalized timestamp.
 
-        The returned expression must accept any SQL value, and return a string.
+        The returned expression must accept any SQL datetime/timestamp, and return a string.
+
+        Date format: "YYYY-MM-DD HH:mm:SS.FFFFFF"
+
+        Precision of dates should be rounded up/down according to coltype.rounds
+        """
+        ...
 
-        - Dates are expected in the format:
-            "YYYY-MM-DD HH:mm:SS.FFFFFF"
+    @abstractmethod
+    def normalize_number(self, value: str, coltype: ColType) -> str:
+        """Creates an SQL expression, that converts 'value' to a normalized number.
 
-            Rounded up/down according to coltype.rounds
+        The returned expression must accept any SQL int/numeric/float, and return a string.
 
         - Floats/Decimals are expected in the format
             "I.P"
@@ -191,14 +203,31 @@ def normalize_value_by_type(value: str, coltype: ColType) -> str:
             and must be at least one digit (0).
             P is the fractional digits, the amount of which is specified with
             coltype.precision. Trailing zeroes may be necessary.
+            If P is 0, the dot is omitted.
 
             Note: This precision is different than the one used by databases. For decimals,
-            it's the same as "numeric_scale", and for floats, who use binary precision,
-            it can be calculated as log10(2**p)
+            it's the same as ``numeric_scale``, and for floats, who use binary precision,
+            it can be calculated as ``log10(2**numeric_precision)``.
+        """
+        ...
+
+    def normalize_value_by_type(self, value: str, coltype: ColType) -> str:
+        """Creates an SQL expression, that converts 'value' to a normalized representation.
+
+        The returned expression must accept any SQL value, and return a string.
+
+        The default implementation dispatches to a method according to ``coltype``:
 
+            TemporalType -> normalize_timestamp()
+            NumericType  -> normalize_number()
+            -else-       -> to_string()
 
         """
-        ...
+        if isinstance(coltype, TemporalType):
+            return self.normalize_timestamp(value, coltype)
+        elif isinstance(coltype, NumericType):
+            return self.normalize_number(value, coltype)
+        return self.to_string(f"{value}")
 
 
 class Database(AbstractDatabase):
@@ -209,8 +238,8 @@ class Database(AbstractDatabase):
     Instanciated using :meth:`~data_diff.connect_to_uri`
     """
 
-    DATETIME_TYPES = NotImplemented
-    default_schema = NotImplemented
+    DATETIME_TYPES = {}
+    default_schema = None
 
     def query(self, sql_ast: SqlOrStr, res_type: type):
         "Query the given SQL code/AST, and attempt to convert the result to type 'res_type'"
@@ -306,13 +335,15 @@ def query_table_schema(self, path: DbPath) -> Dict[str, ColType]:
 
     def _normalize_table_path(self, path: DbPath) -> DbPath:
         if len(path) == 1:
-            return self.default_schema, path[0]
-        elif len(path) == 2:
-            return path
+            if self.default_schema:
+                return self.default_schema, path[0]
+        elif len(path) != 2:
+            raise ValueError(
+                f"{self.__class__.__name__}: Bad table path for {self}: '{'.'.join(path)}'. Expected form: schema.table"
+            )
+
+        return path
 
-        raise ValueError(
-            f"{self.__class__.__name__}: Bad table path for {self}: '{'.'.join(path)}'. Expected form: schema.table"
-        )
 
     def parse_table_name(self, name: str) -> DbPath:
         return parse_table_name(name)
@@ -408,27 +439,16 @@ def md5_to_int(self, s: str) -> str:
     def to_string(self, s: str):
         return f"{s}::varchar"
 
-    def normalize_value_by_type(self, value: str, coltype: ColType) -> str:
-        if isinstance(coltype, TemporalType):
-            # if coltype.precision == 0:
-            #     return f"to_char({value}::timestamp(0), 'YYYY-mm-dd HH24:MI:SS')"
-            # if coltype.precision == 3:
-            #     return f"to_char({value}, 'YYYY-mm-dd HH24:MI:SS.US')"
-            # elif coltype.precision == 6:
-            # return f"to_char({value}::timestamp({coltype.precision}), 'YYYY-mm-dd HH24:MI:SS.US')"
-            # else:
-            #     # Postgres/Redshift doesn't support arbitrary precision
-            #     raise TypeError(f"Bad precision for {type(self).__name__}: {coltype})")
-            if coltype.rounds:
-                return f"to_char({value}::timestamp({coltype.precision}), 'YYYY-mm-dd HH24:MI:SS.US')"
-            else:
-                timestamp6 = f"to_char({value}::timestamp(6), 'YYYY-mm-dd HH24:MI:SS.US')"
-                return f"RPAD(LEFT({timestamp6}, {TIMESTAMP_PRECISION_POS+coltype.precision}), {TIMESTAMP_PRECISION_POS+6}, '0')"
 
-        elif isinstance(coltype, NumericType):
-            value = f"{value}::decimal(38, {coltype.precision})"
+    def normalize_timestamp(self, value: str, coltype: ColType) -> str:
+        if coltype.rounds:
+            return f"to_char({value}::timestamp({coltype.precision}), 'YYYY-mm-dd HH24:MI:SS.US')"
 
-        return self.to_string(f"{value}")
+        timestamp6 = f"to_char({value}::timestamp(6), 'YYYY-mm-dd HH24:MI:SS.US')"
+        return f"RPAD(LEFT({timestamp6}, {TIMESTAMP_PRECISION_POS+coltype.precision}), {TIMESTAMP_PRECISION_POS+6}, '0')"
+
+    def normalize_number(self, value: str, coltype: ColType) -> str:
+        return self.to_string(f"{value}::decimal(38, {coltype.precision})")
 
 
 class Presto(Database):
@@ -468,25 +488,19 @@ def _query(self, sql_code: str) -> list:
     def close(self):
         self._conn.close()
 
-    def normalize_value_by_type(self, value: str, coltype: ColType) -> str:
-        if isinstance(coltype, TemporalType):
-            if coltype.rounds:
-                if coltype.precision > 3:
-                    pass
-                s = f"date_format(cast({value} as timestamp(6)), '%Y-%m-%d %H:%i:%S.%f')"
-            else:
-                s = f"date_format(cast({value} as timestamp(6)), '%Y-%m-%d %H:%i:%S.%f')"
-                # datetime = f"date_format(cast({value} as timestamp(6), '%Y-%m-%d %H:%i:%S.%f'))"
-                # datetime = self.to_string(f"cast({value} as datetime(6))")
-
-            return (
-                f"RPAD(RPAD({s}, {TIMESTAMP_PRECISION_POS+coltype.precision}, '.'), {TIMESTAMP_PRECISION_POS+6}, '0')"
-            )
+    def normalize_timestamp(self, value: str, coltype: ColType) -> str:
+        # TODO
+        if coltype.rounds:
+            s = f"date_format(cast({value} as timestamp(6)), '%Y-%m-%d %H:%i:%S.%f')"
+        else:
+            s = f"date_format(cast({value} as timestamp(6)), '%Y-%m-%d %H:%i:%S.%f')"
 
-        elif isinstance(coltype, NumericType):
-            value = f"cast({value} as decimal(38,{coltype.precision}))"
+        return (
+            f"RPAD(RPAD({s}, {TIMESTAMP_PRECISION_POS+coltype.precision}, '.'), {TIMESTAMP_PRECISION_POS+6}, '0')"
+        )
 
-        return self.to_string(value)
+    def normalize_number(self, value: str, coltype: ColType) -> str:
+        return self.to_string(f"cast({value} as decimal(38,{coltype.precision}))")
 
     def select_table_schema(self, path: DbPath) -> str:
         schema, table = self._normalize_table_path(path)
@@ -575,18 +589,16 @@ def md5_to_int(self, s: str) -> str:
     def to_string(self, s: str):
         return f"cast({s} as char)"
 
-    def normalize_value_by_type(self, value: str, coltype: ColType) -> str:
-        if isinstance(coltype, TemporalType):
-            if coltype.rounds:
-                return self.to_string(f"cast( cast({value} as datetime({coltype.precision})) as datetime(6))")
-            else:
-                s = self.to_string(f"cast({value} as datetime(6))")
-                return f"RPAD(RPAD({s}, {TIMESTAMP_PRECISION_POS+coltype.precision}, '.'), {TIMESTAMP_PRECISION_POS+6}, '0')"
+    def normalize_timestamp(self, value: str, coltype: ColType) -> str:
+        if coltype.rounds:
+            return self.to_string(f"cast( cast({value} as datetime({coltype.precision})) as datetime(6))")
 
-        elif isinstance(coltype, NumericType):
-            value = f"cast({value} as decimal(38,{coltype.precision}))"
+        s = self.to_string(f"cast({value} as datetime(6))")
+        return f"RPAD(RPAD({s}, {TIMESTAMP_PRECISION_POS+coltype.precision}, '.'), {TIMESTAMP_PRECISION_POS+6}, '0')"
+
+    def normalize_number(self, value: str, coltype: ColType) -> str:
+        return self.to_string(f"cast({value} as decimal(38, {coltype.precision}))")
 
-        return self.to_string(f"{value}")
 
 
 class Oracle(ThreadedDatabase):
@@ -631,16 +643,15 @@ def select_table_schema(self, path: DbPath) -> str:
             f" FROM USER_TAB_COLUMNS WHERE table_name = '{table.upper()}'"
         )
 
-    def normalize_value_by_type(self, value: str, coltype: ColType) -> str:
-        if isinstance(coltype, TemporalType):
-            return f"to_char(cast({value} as timestamp({coltype.precision})), 'YYYY-MM-DD HH24:MI:SS.FF6')"
-        elif isinstance(coltype, NumericType):
-            # FM999.9990
-            format_str = "FM" + "9" * (38 - coltype.precision)
-            if coltype.precision:
-                format_str += "0." + "9" * (coltype.precision - 1) + "0"
-            return f"to_char({value}, '{format_str}')"
-        return self.to_string(f"{value}")
+    def normalize_timestamp(self, value: str, coltype: ColType) -> str:
+        return f"to_char(cast({value} as timestamp({coltype.precision})), 'YYYY-MM-DD HH24:MI:SS.FF6')"
+
+    def normalize_number(self, value: str, coltype: ColType) -> str:
+        # FM999.9990
+        format_str = "FM" + "9" * (38 - coltype.precision)
+        if coltype.precision:
+            format_str += "0." + "9" * (coltype.precision - 1) + "0"
+        return f"to_char({value}, '{format_str}')"
 
     def _parse_type(
         self, type_repr: str, datetime_precision: int = None, numeric_precision: int = None, numeric_scale: int = None
@@ -691,27 +702,33 @@ class Redshift(Postgres):
     def md5_to_int(self, s: str) -> str:
         return f"strtol(substring(md5({s}), {1+MD5_HEXDIGITS-CHECKSUM_HEXDIGITS}), 16)::decimal(38)"
 
-    def normalize_value_by_type(self, value: str, coltype: ColType) -> str:
-        if isinstance(coltype, TemporalType):
-            if coltype.rounds:
-                timestamp = f"{value}::timestamp(6)"
-                # Get seconds since epoch. Redshift doesn't support milli- or micro-seconds.
-                secs = f"timestamp 'epoch' + round(extract(epoch from {timestamp})::decimal(38)"
-                # Get the milliseconds from timestamp.
-                ms = f"extract(ms from {timestamp})"
-                # Get the microseconds from timestamp, without the milliseconds!
-                us = f"extract(us from {timestamp})"
-                # epoch = Total time since epoch in microseconds.
-                epoch = f"{secs}*1000000 + {ms}*1000 + {us}"
-                timestamp6 = f"to_char({epoch}, -6+{coltype.precision}) * interval '0.000001 seconds', 'YYYY-mm-dd HH24:MI:SS.US')"
-            else:
-                timestamp6 = f"to_char({value}::timestamp(6), 'YYYY-mm-dd HH24:MI:SS.US')"
-            return f"RPAD(LEFT({timestamp6}, {TIMESTAMP_PRECISION_POS+coltype.precision}), {TIMESTAMP_PRECISION_POS+6}, '0')"
+    def normalize_timestamp(self, value: str, coltype: ColType) -> str:
+        if coltype.rounds:
+            timestamp = f"{value}::timestamp(6)"
+            # Get seconds since epoch. Redshift doesn't support milli- or micro-seconds.
+            secs = f"timestamp 'epoch' + round(extract(epoch from {timestamp})::decimal(38)"
+            # Get the milliseconds from timestamp.
+            ms = f"extract(ms from {timestamp})"
+            # Get the microseconds from timestamp, without the milliseconds!
+            us = f"extract(us from {timestamp})"
+            # epoch = Total time since epoch in microseconds.
+            epoch = f"{secs}*1000000 + {ms}*1000 + {us}"
+            timestamp6 = f"to_char({epoch}, -6+{coltype.precision}) * interval '0.000001 seconds', 'YYYY-mm-dd HH24:MI:SS.US')"
+        else:
+            timestamp6 = f"to_char({value}::timestamp(6), 'YYYY-mm-dd HH24:MI:SS.US')"
+        return f"RPAD(LEFT({timestamp6}, {TIMESTAMP_PRECISION_POS+coltype.precision}), {TIMESTAMP_PRECISION_POS+6}, '0')"
+
+    def normalize_number(self, value: str, coltype: ColType) -> str:
+        return self.to_string(f"{value}::decimal(38,{coltype.precision})")
 
-        elif isinstance(coltype, NumericType):
-            value = f"{value}::decimal(38,{coltype.precision})"
 
-        return self.to_string(f"{value}")
+    def select_table_schema(self, path: DbPath) -> str:
+        schema, table = self._normalize_table_path(path)
+
+        return (
+            "SELECT column_name, data_type, datetime_precision, numeric_precision, numeric_scale FROM information_schema.columns "
+            f"WHERE table_name = '{table.lower()}' AND table_schema = '{schema.lower()}'"
+        )
 
 
 class MsSQL(ThreadedDatabase):
@@ -803,27 +820,23 @@ def select_table_schema(self, path: DbPath) -> str:
             f"WHERE table_name = '{table}' AND table_schema = '{schema}'"
         )
 
-    def normalize_value_by_type(self, value: str, coltype: ColType) -> str:
-        if isinstance(coltype, TemporalType):
-            if coltype.rounds:
-                timestamp = f"timestamp_micros(cast(round(unix_micros(cast({value} as timestamp))/1000000, {coltype.precision})*1000000 as int))"
-                return f"FORMAT_TIMESTAMP('%F %H:%M:%E6S', {timestamp})"
-            else:
-                if coltype.precision == 0:
-                    return f"FORMAT_TIMESTAMP('%F %H:%M:%S.000000, {value})"
-                elif coltype.precision == 6:
-                    return f"FORMAT_TIMESTAMP('%F %H:%M:%E6S', {value})"
+    def normalize_timestamp(self, value: str, coltype: ColType) -> str:
+        if coltype.rounds:
+            timestamp = f"timestamp_micros(cast(round(unix_micros(cast({value} as timestamp))/1000000, {coltype.precision})*1000000 as int))"
+            return f"FORMAT_TIMESTAMP('%F %H:%M:%E6S', {timestamp})"
 
-                timestamp6 = f"FORMAT_TIMESTAMP('%F %H:%M:%E6S', {value})"
-                return f"RPAD(LEFT({timestamp6}, {TIMESTAMP_PRECISION_POS+coltype.precision}), {TIMESTAMP_PRECISION_POS+6}, '0')"
-        elif isinstance(coltype, Integer):
-            pass
+        if coltype.precision == 0:
+            return f"FORMAT_TIMESTAMP('%F %H:%M:%S.000000, {value})"
+        elif coltype.precision == 6:
+            return f"FORMAT_TIMESTAMP('%F %H:%M:%E6S', {value})"
 
-        elif isinstance(coltype, NumericType):
-            # value = f"cast({value} as decimal)"
-            return f"format('%.{coltype.precision}f', ({value}))"
+        timestamp6 = f"FORMAT_TIMESTAMP('%F %H:%M:%E6S', {value})"
+        return f"RPAD(LEFT({timestamp6}, {TIMESTAMP_PRECISION_POS+coltype.precision}), {TIMESTAMP_PRECISION_POS+6}, '0')"
 
-        return self.to_string(f"{value}")
+    def normalize_number(self, value: str, coltype: ColType) -> str:
+        if isinstance(coltype, Integer):
+            return self.to_string(value)
+        return f"format('%.{coltype.precision}f', {value})"
 
     def parse_table_name(self, name: str) -> DbPath:
         path = parse_table_name(name)
@@ -897,19 +910,16 @@ def select_table_schema(self, path: DbPath) -> str:
         schema, table = self._normalize_table_path(path)
         return super().select_table_schema((schema, table))
 
-    def normalize_value_by_type(self, value: str, coltype: ColType) -> str:
-        if isinstance(coltype, TemporalType):
-            if coltype.rounds:
-                timestamp = f"to_timestamp(round(date_part(epoch_nanosecond, {value}::timestamp(9))/1000000000, {coltype.precision}))"
-            else:
-                timestamp = f"cast({value} as timestamp({coltype.precision}))"
-
-            return f"to_char({timestamp}, 'YYYY-MM-DD HH24:MI:SS.FF6')"
+    def normalize_timestamp(self, value: str, coltype: ColType) -> str:
+        if coltype.rounds:
+            timestamp = f"to_timestamp(round(date_part(epoch_nanosecond, {value}::timestamp(9))/1000000000, {coltype.precision}))"
+        else:
+            timestamp = f"cast({value} as timestamp({coltype.precision}))"
 
-        elif isinstance(coltype, NumericType):
-            value = f"cast({value} as decimal(38, {coltype.precision}))"
+        return f"to_char({timestamp}, 'YYYY-MM-DD HH24:MI:SS.FF6')"
 
-        return self.to_string(f"{value}")
+    def normalize_number(self, value: str, coltype: ColType) -> str:
+        return self.to_string(f"cast({value} as decimal(38, {coltype.precision}))")
 
 
 @dataclass