Repository: TobikoData/sqlmesh
Pull request #3498 (2024-12-10 20:18:47)

Fix(bigquery): Pass catalog when checking for clustering key changes


Patch:
diff --git a/sqlmesh/core/engine_adapter/mixins.py b/sqlmesh/core/engine_adapter/mixins.py
index 40668ac5..fb78f4c3 100644
--- a/sqlmesh/core/engine_adapter/mixins.py
+++ b/sqlmesh/core/engine_adapter/mixins.py
@@ -9,6 +9,7 @@ from sqlglot.helper import seq_get
 from sqlmesh.core.engine_adapter.base import EngineAdapter
 from sqlmesh.core.engine_adapter.shared import InsertOverwriteStrategy, SourceQuery
 from sqlmesh.core.node import IntervalUnit
+from sqlmesh.core.dialect import schema_
 from sqlmesh.utils.errors import SQLMeshError
 
 if t.TYPE_CHECKING:
@@ -356,10 +357,15 @@ class ClusteredByMixin(EngineAdapter):
         current_table = exp.to_table(current_table_name)
         target_table = exp.to_table(target_table_name)
 
+        current_table_schema = schema_(current_table.db, catalog=current_table.catalog)
+        target_table_schema = schema_(target_table.db, catalog=target_table.catalog)
+
         current_table_info = seq_get(
-            self.get_data_objects(current_table.db, {current_table.name}), 0
+            self.get_data_objects(current_table_schema, {current_table.name}), 0
+        )
+        target_table_info = seq_get(
+            self.get_data_objects(target_table_schema, {target_table.name}), 0
         )
-        target_table_info = seq_get(self.get_data_objects(target_table.db, {target_table.name}), 0)
 
         if current_table_info and target_table_info:
             if target_table_info.is_clustered:


Test patch:
diff --git a/tests/core/engine_adapter/test_bigquery.py b/tests/core/engine_adapter/test_bigquery.py
index 24b729b6..e9460d15 100644
--- a/tests/core/engine_adapter/test_bigquery.py
+++ b/tests/core/engine_adapter/test_bigquery.py
@@ -18,6 +18,11 @@ from sqlmesh.utils import AttributeDict
 pytestmark = [pytest.mark.bigquery, pytest.mark.engine]
 
 
+@pytest.fixture
+def adapter(make_mocked_engine_adapter: t.Callable) -> BigQueryEngineAdapter:
+    return make_mocked_engine_adapter(BigQueryEngineAdapter)
+
+
 def test_insert_overwrite_by_time_partition_query(
     make_mocked_engine_adapter: t.Callable, mocker: MockerFixture
 ):
@@ -893,3 +898,41 @@ def test_nested_fields_update(make_mocked_engine_adapter: t.Callable, mocker: Mo
         bigquery.SchemaField("details", "STRING", "REPEATED"),
     ]
     assert adapter._build_nested_fields(current_schema, new_nested_fields) == expected
+
+
+def test_get_alter_expressions_includes_catalog(
+    adapter: BigQueryEngineAdapter, mocker: MockerFixture
+):
+    adapter._default_catalog = "test_project"
+
+    columns_mock = mocker.patch(
+        "sqlmesh.core.engine_adapter.bigquery.BigQueryEngineAdapter.columns"
+    )
+    columns_mock.return_value = {
+        "a": exp.DataType.build("int"),
+    }
+
+    get_data_objects_mock = mocker.patch(
+        "sqlmesh.core.engine_adapter.bigquery.BigQueryEngineAdapter.get_data_objects"
+    )
+    get_data_objects_mock.return_value = []
+
+    adapter.get_alter_expressions("catalog1.foo.bar", "catalog2.bar.bing")
+
+    assert get_data_objects_mock.call_count == 2
+
+    schema, tables = get_data_objects_mock.call_args_list[0][0]
+    assert isinstance(schema, exp.Table)
+    assert isinstance(tables, set)
+    assert schema.catalog == "catalog1"
+    assert schema.db == "foo"
+    assert schema.sql(dialect="bigquery") == "catalog1.foo"
+    assert tables == {"bar"}
+
+    schema, tables = get_data_objects_mock.call_args_list[1][0]
+    assert isinstance(schema, exp.Table)
+    assert isinstance(tables, set)
+    assert schema.catalog == "catalog2"
+    assert schema.db == "bar"
+    assert schema.sql(dialect="bigquery") == "catalog2.bar"
+    assert tables == {"bing"}