From 19922bff285aa73530ae319dccdf00514fa1aa66 Mon Sep 17 00:00:00 2001
From: eakmanrq <6326532+eakmanrq@users.noreply.github.com>
Date: Mon, 22 Sep 2025 15:58:00 -0700
Subject: [PATCH] fix: lowercase column names snowflake

---
 sqlmesh/core/engine_adapter/snowflake.py    |  3 +-
 tests/core/engine_adapter/test_snowflake.py | 35 ++++++++++++++++++++-
 2 files changed, 36 insertions(+), 2 deletions(-)

diff --git a/sqlmesh/core/engine_adapter/snowflake.py b/sqlmesh/core/engine_adapter/snowflake.py
index c6b0e71ac3..355fb9719c 100644
--- a/sqlmesh/core/engine_adapter/snowflake.py
+++ b/sqlmesh/core/engine_adapter/snowflake.py
@@ -526,7 +526,8 @@ def _get_data_objects(
                 type=DataObjectType.from_str(row.type),  # type: ignore
                 clustering_key=row.clustering_key,  # type: ignore
             )
-            for row in df.itertuples()
+            # lowercase the column names for cases where Snowflake might return uppercase column names for certain catalogs
+            for row in df.rename(columns={col: col.lower() for col in df.columns}).itertuples()
         ]
 
     def set_current_catalog(self, catalog: str) -> None:
diff --git a/tests/core/engine_adapter/test_snowflake.py b/tests/core/engine_adapter/test_snowflake.py
index 75ce8edbe0..62c4a4f3eb 100644
--- a/tests/core/engine_adapter/test_snowflake.py
+++ b/tests/core/engine_adapter/test_snowflake.py
@@ -7,9 +7,10 @@
 
 import sqlmesh.core.dialect as d
 from sqlmesh.core.dialect import normalize_model_name
+from sqlmesh.core.engine_adapter import SnowflakeEngineAdapter
 from sqlmesh.core.engine_adapter.base import EngineAdapter
+from sqlmesh.core.engine_adapter.shared import DataObjectType
 from sqlmesh.core.model import load_sql_based_model
-from sqlmesh.core.engine_adapter import SnowflakeEngineAdapter
 from sqlmesh.core.model.definition import SqlModel
 from sqlmesh.core.node import IntervalUnit
 from sqlmesh.utils.errors import SQLMeshError
@@ -39,6 +40,38 @@ def test_get_temp_table(mocker: MockerFixture, make_mocked_engine_adapter: t.Cal
     assert value.sql(dialect=adapter.dialect) == '"CATALOG"."DB"."__temp_TEST_TABLE_abcdefgh"'
 
 
+def test_get_data_objects_lowercases_columns(
+    make_mocked_engine_adapter: t.Callable, mocker: MockerFixture
+) -> None:
+    adapter = make_mocked_engine_adapter(SnowflakeEngineAdapter, patch_get_data_objects=False)
+
+    adapter.get_current_catalog = mocker.Mock(return_value="TEST_CATALOG")
+
+    adapter.fetchdf = mocker.Mock(
+        return_value=pd.DataFrame(  # type: ignore[assignment]
+            [
+                {
+                    "CATALOG": "TEST_CATALOG",
+                    "NAME": "MY_TABLE",
+                    "SCHEMA_NAME": "PUBLIC",
+                    "TYPE": "TABLE",
+                    "CLUSTERING_KEY": "ID",
+                }
+            ]
+        )
+    )
+
+    data_objects = adapter._get_data_objects("TEST_CATALOG.PUBLIC")
+
+    assert len(data_objects) == 1
+    data_object = data_objects[0]
+    assert data_object.catalog == "TEST_CATALOG"
+    assert data_object.schema_name == "PUBLIC"
+    assert data_object.name == "MY_TABLE"
+    assert data_object.type == DataObjectType.TABLE
+    assert data_object.clustering_key == "ID"
+
+
 @pytest.mark.parametrize(
     "current_warehouse, current_warehouse_exp, configured_warehouse, configured_warehouse_exp, should_change",
     [