include catalog in table name (#1241)

Author: eakmanrq

sqlmesh/core/snapshot/definition.py

@@ -875,7 +875,14 @@ def _ensure_categorized(self) -> None:
 
 def table_name(physical_schema: str, name: str, version: str, is_temp: bool = False) -> str:
     temp_suffx = "__temp" if is_temp else ""
-    return f"{physical_schema}.{name.replace('.', '__')}__{version}{temp_suffx}"
+    parts = [
+        physical_schema,
+        f"{name.replace('.', '__')}__{version}{temp_suffx}",
+    ]
+    catalog = exp.to_table(name).catalog
+    if catalog:
+        parts.insert(0, catalog)
+    return ".".join(parts)
 
 
 def fingerprint_from_node(

tests/core/test_snapshot.py

@@ -1,4 +1,5 @@
 import json
+import typing as t
 from copy import deepcopy
 from datetime import timedelta
 from pathlib import Path
@@ -509,7 +510,7 @@ def test_stamp(model: Model):
     assert original_fingerprint != stamped_fingerprint
 
 
-def test_table_name(snapshot: Snapshot):
+def test_table_name(snapshot: Snapshot, make_snapshot: t.Callable):
     # Mimic a direct breaking change.
     snapshot.fingerprint = SnapshotFingerprint(
         data_hash="1", metadata_hash="1", parent_data_hash="1"
@@ -574,6 +575,15 @@ def test_table_name(snapshot: Snapshot):
     snapshot.physical_schema_ = "private"
     assert snapshot.table_name_for_mapping(is_dev=True) == "private.name__3078928823"
 
+    fully_qualified_snapshot = make_snapshot(
+        SqlModel(name="catalog.db.table", query=parse_one("select 1, ds"))
+    )
+    fully_qualified_snapshot.categorize_as(SnapshotChangeCategory.BREAKING)
+    assert (
+        fully_qualified_snapshot.table_name(is_dev=False, for_read=False)
+        == "catalog.sqlmesh__db.catalog__db__table__2528031000"
+    )
+
 
 def test_categorize_change_sql(make_snapshot):
     old_snapshot = make_snapshot(SqlModel(name="a", query=parse_one("select 1, ds")))