Fix: Always recreate materialized view

Author: VaggelisD

sqlmesh/core/snapshot/evaluator.py

@@ -1876,6 +1876,9 @@ def insert(
         )
         snapshot = kwargs["snapshot"]
         snapshots = kwargs["snapshots"]
+
+        # We must replace a materialized view if its query has changed or if it depends on a table.
+        # The latter is because if the underlying table is replaced, the materialized view is invalidated in some of the engines.
         if (
             (
                 isinstance(query_or_df, exp.Expression)
@@ -1887,6 +1890,7 @@ def insert(
                     engine_adapter=self.adapter,
                 )
                 == query_or_df
+                and not model.depends_on
             )
             or self.adapter.HAS_VIEW_BINDING
         ) and self.adapter.table_exists(table_name):

tests/core/engine_adapter/integration/test_integration_bigquery.py

@@ -400,3 +400,50 @@ def test_table_diff_table_name_matches_column_name(ctx: TestContext):
 
     assert row_diff.stats["join_count"] == 1
     assert row_diff.full_match_count == 1
+
+
+def test_materialized_view_evaluation(ctx: TestContext, engine_adapter: BigQueryEngineAdapter):
+    model_name = ctx.table("test_tbl")
+    mview_name = ctx.table("test_mview")
+
+    sqlmesh = ctx.create_context()
+
+    sqlmesh.upsert_model(
+        load_sql_based_model(
+            d.parse(
+                f"""
+                MODEL (name {model_name}, kind FULL);
+
+                SELECT 1 AS col
+                """
+            )
+        )
+    )
+
+    sqlmesh.upsert_model(
+        load_sql_based_model(
+            d.parse(
+                f"""
+                MODEL (name {mview_name}, kind VIEW (materialized true));
+
+                SELECT * FROM {model_name}
+                """
+            )
+        )
+    )
+
+    # Case 1: Ensure that plan is successful and we can query the materialized view
+    sqlmesh.plan(auto_apply=True, no_prompts=True)
+
+    df = engine_adapter.fetchdf(f"SELECT * FROM {mview_name.sql(dialect=ctx.dialect)}")
+    assert df["col"][0] == 1
+
+    # Case 2: Ensure that we can change the underlying table and the materialized view is recreated
+    sqlmesh.upsert_model(
+        load_sql_based_model(d.parse(f"""MODEL (name {model_name}, kind FULL); SELECT 2 AS col"""))
+    )
+
+    sqlmesh.plan(auto_apply=True, no_prompts=True)
+
+    df = engine_adapter.fetchdf(f"SELECT * FROM {mview_name.sql(dialect=ctx.dialect)}")
+    assert df["col"][0] == 2

tests/core/test_snapshot_evaluator.py

@@ -516,25 +516,8 @@ def test_evaluate_materialized_view(
         snapshots={},
     )
 
-    adapter_mock.table_exists.assert_called_once_with(snapshot.table_name())
-
-    if view_exists:
-        # Evaluation shouldn't take place because the rendered query hasn't changed
-        # since the last view creation.
-        assert not adapter_mock.create_view.called
-    else:
-        # If the view doesn't exist, it should be created even if the rendered query
-        # hasn't changed since the last view creation.
-        adapter_mock.create_view.assert_called_once_with(
-            snapshot.table_name(),
-            model.render_query(),
-            model.columns_to_types,
-            replace=True,
-            materialized=True,
-            view_properties={},
-            table_description=None,
-            column_descriptions={},
-        )
+    # Ensure that the materialized view is recreated even if it exists
+    assert adapter_mock.create_view.assert_called
 
 
 def test_evaluate_materialized_view_with_partitioned_by_cluster_by(