feat: Add support for column type comparison in schema checks and reports

Author: luisggoncalves

sqlcompare/db/connection.py

@@ -7,7 +7,7 @@
 from typing import Any
 from urllib.parse import parse_qs, urlparse
 
-from sqlalchemy import text
+from sqlalchemy import inspect, text
 from sqlalchemy.engine import Connection, Engine, Result
 from sqlalchemy.exc import SQLAlchemyError
 
@@ -261,6 +261,26 @@ def get_table_columns(self, table_name: str) -> list[str]:
         _, columns = self.query(f"SELECT * FROM {table_name} WHERE 1=0", include_columns=True)
         return columns
 
+    def get_table_column_types(self, table_name: str) -> dict[str, str | None]:
+        """
+        Get column types keyed by the actual column name as stored in the database.
+        """
+        inspector = inspect(self.conn)
+        schema = None
+        table = table_name
+        if "." in table_name:
+            schema, table = table_name.rsplit(".", 1)
+            schema = schema.strip('"')
+            table = table.strip('"')
+
+        columns = inspector.get_columns(table, schema=schema)
+        return {
+            str(column["name"]): (
+                None if column.get("type") is None else str(column["type"]).upper()
+            )
+            for column in columns
+        }
+
     def query(
         self,
         sql: str,

sqlcompare/stats/checks.py

@@ -139,12 +139,22 @@ def run_row_count_check(context: StatsContext, definition: CheckDefinition) -> C
 def run_schema_check(context: StatsContext, definition: CheckDefinition) -> CheckResult:
     previous_only = context.previous_only_columns
     current_only = context.current_only_columns
-    has_differences = bool(previous_only or current_only)
+    type_changes = [
+        (
+            pair.previous_name,
+            pair.previous_type or "unknown",
+            pair.current_type or "unknown",
+        )
+        for pair in context.common_columns
+        if (pair.previous_type or "unknown").upper() != (pair.current_type or "unknown").upper()
+    ]
+    has_differences = bool(previous_only or current_only or type_changes)
     common_count = len(context.common_columns)
     if has_differences:
         summary = (
             f"{len(previous_only)} only in previous, "
             f"{len(current_only)} only in current, "
+            f"{len(type_changes)} type changes, "
             f"{common_count} common columns."
         )
     else:
@@ -157,6 +167,7 @@ def run_schema_check(context: StatsContext, definition: CheckDefinition) -> Chec
         metadata={
             "previous_only_columns": previous_only,
             "current_only_columns": current_only,
+            "type_changes": type_changes,
             "common_count": common_count,
         },
     )

sqlcompare/stats/models.py

@@ -9,6 +9,8 @@ class ColumnPair:
     key: str
     previous_name: str
     current_name: str
+    previous_type: str | None = None
+    current_type: str | None = None
 
 
 @dataclass
@@ -21,6 +23,8 @@ class StatsContext:
     common_columns: list[ColumnPair]
     previous_only_columns: list[str]
     current_only_columns: list[str]
+    previous_column_types: dict[str, str | None]
+    current_column_types: dict[str, str | None]
 
 
 @dataclass

sqlcompare/stats/report.py

@@ -24,12 +24,23 @@ def _status_cell(metric: dict) -> str:
     return f"{previous} -> {current} ({pct_diff})"
 
 
+def _type_status_cell(previous_type: str | None, current_type: str | None) -> str:
+    previous = previous_type or "unknown"
+    current = current_type or "unknown"
+
+    if previous.upper() == current.upper():
+        return f"{current} (no change)"
+
+    return f"{previous} -> {current}"
+
+
 def _build_column_comparison_table(
     context: StatsContext,
     selected_check_names: list[str],
     results_by_name: dict[str, CheckResult],
 ) -> tuple[list[str], list[tuple]]:
     columns = ["COL"]
+    columns.append("TYPE")
     rows: list[tuple] = []
 
     include_nulls = "nulls" in selected_check_names and "nulls" in results_by_name
@@ -57,6 +68,7 @@ def _build_column_comparison_table(
 
     for pair in context.common_columns:
         row = [pair.previous_name]
+        row.append(_type_status_cell(pair.previous_type, pair.current_type))
         similarities: list[float] = []
 
         if row_count_similarity is not None:
@@ -107,12 +119,19 @@ def render_report(
             lines.append(result.summary or "")
             prev_only = result.metadata.get("previous_only_columns", [])
             curr_only = result.metadata.get("current_only_columns", [])
+            type_changes = result.metadata.get("type_changes", [])
             if prev_only:
                 lines.append("Only in previous:")
                 lines.extend(f"  - {col}" for col in prev_only)
             if curr_only:
                 lines.append("Only in current:")
                 lines.extend(f"  - {col}" for col in curr_only)
+            if type_changes:
+                lines.append("Type changes:")
+                lines.extend(
+                    f"  - {column}: {previous_type} -> {current_type}"
+                    for column, previous_type, current_type in type_changes
+                )
             continue
 
         if result.skipped_reason:

sqlcompare/stats/runner.py

@@ -64,6 +64,8 @@ def _resolve_checks(checks: str | None) -> list[str]:
 def _build_context(db: DBConnection, previous_name: str, current_name: str) -> StatsContext:
     previous_columns = db.get_table_columns(previous_name)
     current_columns = db.get_table_columns(current_name)
+    previous_column_types = db.get_table_column_types(previous_name)
+    current_column_types = db.get_table_column_types(current_name)
     previous_map = {column.upper(): column for column in previous_columns}
     current_map = {column.upper(): column for column in current_columns}
 
@@ -72,6 +74,8 @@ def _build_context(db: DBConnection, previous_name: str, current_name: str) -> S
             key=key,
             previous_name=previous_map[key],
             current_name=current_map[key],
+            previous_type=previous_column_types.get(previous_map[key]),
+            current_type=current_column_types.get(current_map[key]),
         )
         for key in previous_map
         if key in current_map
@@ -92,6 +96,8 @@ def _build_context(db: DBConnection, previous_name: str, current_name: str) -> S
         common_columns=common_columns,
         previous_only_columns=previous_only_columns,
         current_only_columns=current_only_columns,
+        previous_column_types=previous_column_types,
+        current_column_types=current_column_types,
     )
 
 

tests/test_stats_comparison.py

@@ -42,6 +42,7 @@ def test_table_command_with_stats(tmp_path, monkeypatch) -> None:
     assert "Previous  Current  Diff % Diff" in result.output
     assert "        2        2     0   0.0%" in result.output
     assert "COL" in result.output
+    assert "TYPE" in result.output
     assert "MATCH" in result.output
     assert "0 (no change)" in result.output
     assert "Schemas match on column names (3 common columns)." in result.output
@@ -71,8 +72,10 @@ def test_table_command_with_stats_from_files() -> None:
     assert "Column comparison:" in result.output
     assert "NULL" in result.output
     assert "DUP" in result.output
+    assert "TYPE" in result.output
     assert "MATCH" in result.output
     assert "notes" in result.output
+    assert "VARCHAR (no change)" in result.output
     assert "2 -> 1 (-50%)" in result.output
     assert "0 (no change)" in result.output
 
@@ -223,6 +226,25 @@ def test_stats_command_reports_schema_differences(tmp_path) -> None:
     assert "Column comparison:" in result.output
 
 
+def test_stats_command_reports_type_changes_in_column_comparison(tmp_path) -> None:
+    previous = tmp_path / "previous.csv"
+    current = tmp_path / "current.csv"
+    previous.write_text("id,code\n1,100\n2,200\n", encoding="utf-8")
+    current.write_text("id,code\n1,A100\n2,B200\n", encoding="utf-8")
+    runner = CliRunner()
+
+    result = runner.invoke(app, ["run", "stats", str(previous), str(current)])
+
+    assert result.exit_code == 0, result.output
+    assert "Schema differences:" in result.output
+    assert "Type changes:" in result.output
+    assert "code: BIGINT -> VARCHAR" in result.output
+    assert "Column comparison:" in result.output
+    assert "TYPE" in result.output
+    assert "BIGINT -> VARCHAR" in result.output
+    assert "BIGINT (no change)" in result.output
+
+
 def test_stats_command_skips_column_checks_when_no_common_columns(tmp_path) -> None:
     previous = tmp_path / "previous.csv"
     current = tmp_path / "current.csv"