Fix: Always force recreation of materialized views

Author: VaggelisD

sqlmesh/core/snapshot/evaluator.py

@@ -2297,7 +2297,7 @@ def insert(
             table_name,
             query_or_df,
             model.columns_to_types,
-            replace=not self.adapter.HAS_VIEW_BINDING,
+            replace=snapshot.is_materialized_view or not self.adapter.HAS_VIEW_BINDING,
             materialized=self._is_materialized_view(model),
             view_properties=kwargs.get("physical_properties", model.physical_properties),
             table_description=model.description,

tests/core/engine_adapter/integration/test_integration.py

@@ -3731,3 +3731,60 @@ def _set_config(_gateway: str, config: Config) -> None:
         "incremental_model",
         "seed_model",
     ]
+
+
+def test_materialized_view_evaluation(ctx: TestContext):
+    adapter = ctx.engine_adapter
+    dialect = ctx.dialect
+    if not adapter.SUPPORTS_MATERIALIZED_VIEWS:
+        pytest.skip(f"Skipping engine {dialect} as it does not support materialized views")
+
+    elif dialect == "snowflake":
+        pytest.skip(
+            f"Skipping Snowflake as it requires an enterprise account for materialized views"
+        )
+
+    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 = adapter.fetchdf(f"SELECT * FROM {mview_name.sql(dialect=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 = adapter.fetchdf(f"SELECT * FROM {mview_name.sql(dialect=dialect)}")
+    assert df["col"][0] == 2

tests/core/engine_adapter/integration/test_integration_bigquery.py

@@ -441,53 +441,6 @@ def test_table_diff_table_name_matches_column_name(ctx: TestContext):
     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
-
-
 def test_correlation_id_in_job_labels(ctx: TestContext):
     model_name = ctx.table("test")