fix(snowflake): keep quoted and unquoted catalogs apart in last modified lookup

Group the tables by the catalog identifier instead of its name. In
Snowflake an unquoted raw is the database RAW while a quoted "raw" is a
different database, so grouping by name sent both to whichever
information schema came first.

Signed-off-by: Dhruv Gupta <todhruvg@gmail.com>

Author: DG47

sqlmesh/core/engine_adapter/snowflake.py

@@ -758,21 +758,19 @@ def get_table_last_modified_ts(self, table_names: t.List[TableName]) -> t.List[i
         from sqlmesh.utils.date import to_timestamp
 
         # INFORMATION_SCHEMA only contains the objects of the database it belongs to, so the tables
-        # are grouped by database and each group is looked up in its own database's INFORMATION_SCHEMA
-        catalogs_to_tables: t.Dict[str, t.List[exp.Table]] = {}
+        # are grouped by database and each group is looked up in its own database's INFORMATION_SCHEMA.
+        # The groups are keyed on the catalog identifier rather than its name, because quoting matters
+        # in Snowflake: unquoted raw is the database RAW, while quoted "raw" is a different database
+        catalogs_to_tables: t.Dict[t.Optional[exp.Identifier], t.List[exp.Table]] = {}
         for table_name in table_names:
             table = exp.to_table(table_name)
-            catalogs_to_tables.setdefault(table.catalog, []).append(table)
+            catalogs_to_tables.setdefault(table.args.get("catalog"), []).append(table)
 
         results = []
-        for tables in catalogs_to_tables.values():
+        for catalog, tables in catalogs_to_tables.items():
             query = (
                 exp.select("LAST_ALTERED")
-                .from_(
-                    exp.table_(
-                        "TABLES", db="INFORMATION_SCHEMA", catalog=tables[0].args.get("catalog")
-                    )
-                )
+                .from_(exp.table_("TABLES", db="INFORMATION_SCHEMA", catalog=catalog))
                 .where(
                     exp.or_(
                         *(

tests/core/engine_adapter/test_snowflake.py

@@ -205,6 +205,29 @@ def test_session(
         assert to_sql_calls(adapter) == expected_exception_calls
 
 
+def test_get_table_last_modified_ts_quoted_and_unquoted_catalogs(
+    snowflake_mocked_engine_adapter: SnowflakeEngineAdapter,
+) -> None:
+    # unquoted raw resolves to the database RAW, quoted "raw" is a different database,
+    # so the two must be looked up in two different information schemas
+    adapter = snowflake_mocked_engine_adapter
+    adapter.cursor.fetchall.side_effect = [
+        [(datetime(2025, 1, 1, tzinfo=timezone.utc),)],
+        [(datetime(2025, 1, 2, tzinfo=timezone.utc),)],
+    ]
+
+    result = adapter.get_table_last_modified_ts(['"raw".shopify.orders', "raw.shopify.customers"])
+
+    assert result == [
+        to_timestamp(datetime(2025, 1, 1, tzinfo=timezone.utc)),
+        to_timestamp(datetime(2025, 1, 2, tzinfo=timezone.utc)),
+    ]
+    assert to_sql_calls(adapter) == [
+        """SELECT LAST_ALTERED FROM "raw".INFORMATION_SCHEMA.TABLES WHERE TABLE_NAME = 'orders' AND TABLE_SCHEMA = 'shopify'""",
+        """SELECT LAST_ALTERED FROM raw.INFORMATION_SCHEMA.TABLES WHERE TABLE_NAME = 'customers' AND TABLE_SCHEMA = 'shopify'""",
+    ]
+
+
 def test_comments(make_mocked_engine_adapter: t.Callable, mocker: MockerFixture):
     adapter = make_mocked_engine_adapter(SnowflakeEngineAdapter)