Fix: Only parse relevant columns. Only warn on relevant columns.

Author: erezsh

data_diff/database.py

@@ -5,7 +5,7 @@
 from abc import ABC, abstractmethod
 from runtype import dataclass
 import logging
-from typing import Tuple, Optional, List
+from typing import Sequence, Tuple, Optional, List
 from concurrent.futures import ThreadPoolExecutor
 import threading
 from typing import Dict
@@ -131,10 +131,6 @@ 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
@@ -163,7 +159,7 @@ def select_table_schema(self, path: DbPath) -> str:
         ...
 
     @abstractmethod
-    def query_table_schema(self, path: DbPath) -> Dict[str, ColType]:
+    def query_table_schema(self, path: DbPath, filter_columns: Optional[Sequence[str]] = None) -> Dict[str, ColType]:
         "Query the table for its schema for table in 'path', and return {column: type}"
         ...
 
@@ -241,6 +237,10 @@ class Database(AbstractDatabase):
     DATETIME_TYPES = {}
     default_schema = None
 
+    @property
+    def name(self):
+        return type(self).__name__
+
     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'"
 
@@ -321,12 +321,16 @@ def select_table_schema(self, path: DbPath) -> str:
             f"WHERE table_name = '{table}' AND table_schema = '{schema}'"
         )
 
-    def query_table_schema(self, path: DbPath) -> Dict[str, ColType]:
+    def query_table_schema(self, path: DbPath, filter_columns: Optional[Sequence[str]] = None) -> Dict[str, ColType]:
         rows = self.query(self.select_table_schema(path), list)
         if not rows:
-            raise RuntimeError(f"{self.__class__.__name__}: Table '{'.'.join(path)}' does not exist, or has no columns")
+            raise RuntimeError(f"{self.name}: Table '{'.'.join(path)}' does not exist, or has no columns")
+
+        if filter_columns is not None:
+            accept = {i.lower() for i in filter_columns}
+            rows = [r for r in rows if r[0].lower() in accept]
 
-        # Return a dict of form {name: type} after canonizaation
+        # Return a dict of form {name: type} after normalization
         return {row[0]: self._parse_type(*row[1:]) for row in rows}
 
     # @lru_cache()
@@ -339,7 +343,7 @@ def _normalize_table_path(self, path: DbPath) -> DbPath:
                 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"
+                f"{self.name}: Bad table path for {self}: '{'.'.join(path)}'. Expected form: schema.table"
             )
 
         return path
@@ -407,6 +411,7 @@ class Postgres(ThreadedDatabase):
         "decimal": Decimal,
         "integer": Integer,
         "numeric": Decimal,
+        "bigint": Integer,
     }
     ROUNDS_ON_PREC_LOSS = True
 

data_diff/diff_tables.py

@@ -12,7 +12,7 @@
 from runtype import dataclass
 
 from .sql import Select, Checksum, Compare, DbPath, DbKey, DbTime, Count, TableName, Time, Min, Max
-from .database import Database, NumericType, PrecisionType, ColType
+from .database import Database, NumericType, PrecisionType, ColType, UnknownColType
 
 logger = logging.getLogger("diff_tables")
 
@@ -142,7 +142,8 @@ def with_schema(self) -> "TableSegment":
         "Queries the table schema from the database, and returns a new instance of TableSegmentWithSchema."
         if self._schema:
             return self
-        schema = self.database.query_table_schema(self.table_path)
+
+        schema = self.database.query_table_schema(self.table_path, self._relevant_columns)
         if self.case_sensitive:
             schema = Schema_CaseSensitive(schema)
         else:
@@ -381,6 +382,13 @@ def _validate_and_adjust_columns(self, table1, table2):
                 table1._schema[c] = col1.replace(precision=lowest.precision)
                 table2._schema[c] = col2.replace(precision=lowest.precision)
 
+        for t in [table1, table2]:
+            for c in t._relevant_columns:
+                ctype = t._schema[c]
+                if isinstance(ctype, UnknownColType):
+                    logger.warn(f"[{t.database.name}] Column '{c}' of type '{ctype.text}' has no compatibility handling. "
+                                "If encoding/formatting differs between databases, it may result in false positives.")
+
     def _bisect_and_diff_tables(self, table1, table2, level=0, max_rows=None):
         assert table1.is_bounded and table2.is_bounded