fix(snowflake): look up last modified timestamps in each table's database

SnowflakeEngineAdapter.get_table_last_modified_ts queried the unqualified
INFORMATION_SCHEMA.TABLES view, which only contains objects from the
connection's current database. Tables in any other database were never
found, so the freshness signal raised MissingSourceError for models that
read external tables from another database.

Group the tables by database and query each database's own
INFORMATION_SCHEMA.TABLES. The query is now built with sqlglot instead of
string formatting, so table and schema names are escaped as literals.

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

Author: DG47

sqlmesh/core/engine_adapter/snowflake.py

@@ -757,14 +757,34 @@ def close(self) -> t.Any:
     def get_table_last_modified_ts(self, table_names: t.List[TableName]) -> t.List[int]:
         from sqlmesh.utils.date import to_timestamp
 
-        num_tables = len(table_names)
-
-        query = "SELECT LAST_ALTERED FROM INFORMATION_SCHEMA.TABLES WHERE"
-        for i, table_name in enumerate(table_names):
+        # 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]] = {}
+        for table_name in table_names:
             table = exp.to_table(table_name)
-            query += f"""(TABLE_NAME = '{table.name}' AND TABLE_SCHEMA = '{table.db}' AND TABLE_CATALOG = '{table.catalog}')"""
-            if i < num_tables - 1:
-                query += " OR "
+            catalogs_to_tables.setdefault(table.catalog, []).append(table)
+
+        results = []
+        for tables in catalogs_to_tables.values():
+            query = (
+                exp.select("LAST_ALTERED")
+                .from_(
+                    exp.table_(
+                        "TABLES", db="INFORMATION_SCHEMA", catalog=tables[0].args.get("catalog")
+                    )
+                )
+                .where(
+                    exp.or_(
+                        *(
+                            exp.and_(
+                                exp.column("TABLE_NAME").eq(exp.Literal.string(table.name)),
+                                exp.column("TABLE_SCHEMA").eq(exp.Literal.string(table.db)),
+                            )
+                            for table in tables
+                        )
+                    )
+                )
+            )
+            results.extend(self.fetchall(query))
 
-        result = self.fetchall(query)
-        return [to_timestamp(row[0]) for row in result]
+        return [to_timestamp(row[0]) for row in results]

tests/core/engine_adapter/test_snowflake.py

@@ -1,4 +1,5 @@
 import typing as t
+from datetime import datetime, timezone
 
 import pandas as pd  # noqa: TID253
 import pytest
@@ -14,6 +15,7 @@
 from sqlmesh.core.model import load_sql_based_model
 from sqlmesh.core.model.definition import SqlModel
 from sqlmesh.core.node import IntervalUnit
+from sqlmesh.utils.date import to_timestamp
 from sqlmesh.utils.errors import SQLMeshError
 from sqlmesh.utils import optional_import
 from tests.core.engine_adapter import to_sql_calls
@@ -73,6 +75,38 @@ def test_get_data_objects_lowercases_columns(
     assert data_object.clustering_key == "ID"
 
 
+def test_get_table_last_modified_ts_multiple_catalogs(
+    snowflake_mocked_engine_adapter: SnowflakeEngineAdapter,
+) -> None:
+    adapter = snowflake_mocked_engine_adapter
+    adapter.cursor.fetchall.side_effect = [
+        [
+            (datetime(2025, 1, 1, tzinfo=timezone.utc),),
+            (datetime(2025, 1, 2, tzinfo=timezone.utc),),
+        ],
+        [(datetime(2025, 1, 3, tzinfo=timezone.utc),)],
+    ]
+
+    result = adapter.get_table_last_modified_ts(
+        [
+            '"ANALYTICS"."STAGING"."ORDERS"',
+            '"RAW"."SHOPIFY"."ORDERS"',
+            '"ANALYTICS"."STAGING"."CUSTOMERS"',
+        ]
+    )
+
+    # INFORMATION_SCHEMA only covers the database it belongs to, so each database is queried separately
+    assert to_sql_calls(adapter) == [
+        """SELECT LAST_ALTERED FROM "ANALYTICS".INFORMATION_SCHEMA.TABLES WHERE (TABLE_NAME = 'ORDERS' AND TABLE_SCHEMA = 'STAGING') OR (TABLE_NAME = 'CUSTOMERS' AND TABLE_SCHEMA = 'STAGING')""",
+        """SELECT LAST_ALTERED FROM "RAW".INFORMATION_SCHEMA.TABLES WHERE TABLE_NAME = 'ORDERS' AND TABLE_SCHEMA = 'SHOPIFY'""",
+    ]
+    assert result == [
+        to_timestamp("2025-01-01"),
+        to_timestamp("2025-01-02"),
+        to_timestamp("2025-01-03"),
+    ]
+
+
 @pytest.mark.parametrize(
     "current_warehouse, current_warehouse_exp, configured_warehouse, configured_warehouse_exp, should_change",
     [