feat: Add support for Snowflake table column type retrieval using DESCRIBE TABLE

Author: luisggoncalves

sqlcompare/db/connection.py

@@ -265,6 +265,29 @@ 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.
         """
+        try:
+            dialect_name = self._engine.dialect.name.lower() if self._engine else None
+        except Exception:
+            dialect_name = None
+
+        # SQLAlchemy's Snowflake inspector mis-parses three-part names by combining
+        # the current database with the provided database.schema prefix. DESCRIBE
+        # TABLE accepts the fully qualified name directly and preserves identifier case.
+        if dialect_name == "snowflake":
+            try:
+                result = self.query(f"DESCRIBE TABLE {table_name}")
+                if result:
+                    return {
+                        str(row[0]): (
+                            None
+                            if len(row) < 2 or row[1] is None
+                            else str(row[1]).upper()
+                        )
+                        for row in result
+                    }
+            except Exception:
+                pass
+
         inspector = inspect(self.conn)
         schema = None
         table = table_name

tests/test_db_connection.py

@@ -0,0 +1,36 @@
+from __future__ import annotations
+
+from types import SimpleNamespace
+
+from sqlcompare.db import DBConnection
+import sqlcompare.db.connection as connection_module
+
+
+def test_get_table_column_types_snowflake_uses_describe_for_fully_qualified_name(
+    monkeypatch,
+) -> None:
+    db = DBConnection("snowflake://example")
+    db._engine = SimpleNamespace(dialect=SimpleNamespace(name="snowflake"))
+
+    captured: dict[str, str] = {}
+
+    def fake_query(sql: str, *args, **kwargs):
+        captured["sql"] = sql
+        return [("ID", "NUMBER(38,0)"), ("NAME", "VARCHAR(16777216)")]
+
+    monkeypatch.setattr(db, "query", fake_query)
+
+    def fail_inspect(_conn):
+        raise AssertionError("inspect() should not be used for Snowflake")
+
+    monkeypatch.setattr(connection_module, "inspect", fail_inspect)
+
+    result = db.get_table_column_types(
+        "LH_PROD_DATA_SNOWFLAKE.salesforce_derived.account_augmented"
+    )
+
+    assert (
+        captured["sql"]
+        == "DESCRIBE TABLE LH_PROD_DATA_SNOWFLAKE.salesforce_derived.account_augmented"
+    )
+    assert result == {"ID": "NUMBER(38,0)", "NAME": "VARCHAR(16777216)"}