Force mview creation again in RW

Author: VaggelisD

.circleci/continue_config.yml

@@ -310,10 +310,10 @@ workflows:
                 - athena                
                 - fabric
                 - gcp-postgres
-          filters:
-            branches:
-              only:
-                - main
+          # filters:
+          #   branches:
+          #     only:
+          #       - main
       - ui_style
       - ui_test
       - vscode_test

sqlmesh/core/snapshot/evaluator.py

@@ -2282,11 +2282,14 @@ def insert(
         render_kwargs: t.Dict[str, t.Any],
         **kwargs: t.Any,
     ) -> None:
-        # For all engines that support materialized views (except RisingWave) we recreate it.
-        # This is because case if upstream tables were recreated (e.g FULL models), the mview would be invalidated as well.
-        # RisingWave is an exception as it doesn't support CREATE OR REPLACE, so upstream models don't recreate their physical tables.
-        # Standard views still follow the same logic depending on VIEW_BINDING flag.
-        must_recreate_view = not self.adapter.HAS_VIEW_BINDING
+        # We should always recreate materialized views across supported engines (Snowflake, BigQuery etc) because
+        # if upstream tables were recreated (e.g FULL models), the MVs would be silently invalidated.
+        # The only exception to that rule is RisingWave, which doesn't support CREATE OR REPLACE, so upstream 
+        # models don't recreate their physical tables for the MVs to be invalidated.
+        # However, even for RW we still want to recreate MVs to avoid stale references  
+        is_materialized_view = self._is_materialized_view(model)
+        must_recreate_view = not self.adapter.HAS_VIEW_BINDING or is_materialized_view
+
         if self.adapter.table_exists(table_name) and not must_recreate_view:
             logger.info("Skipping creation of the view '%s'", table_name)
             return

tests/core/engine_adapter/integration/test_integration.py

@@ -3745,7 +3745,7 @@ def test_materialized_view_evaluation(ctx: TestContext, mocker: MockerFixture):
         pytest.skip(f"Skipping {dialect} as they're not enabled on standard accounts")
 
     model_name = ctx.table("test_tbl")
-    mview_name = ctx.table("test_mview")
+    mview_name = ctx.table("test_mview1")
 
     sqlmesh = ctx.create_context()
 
@@ -3772,12 +3772,15 @@ def test_materialized_view_evaluation(ctx: TestContext, mocker: MockerFixture):
             )
         )
     )
+    
+    def _assert_mview_value(value: int):
+        df = adapter.fetchdf(f"SELECT * FROM {mview_name.sql(dialect=dialect)}")
+        assert df["col"][0] == value
 
     # 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
+    _assert_mview_value(value=1)
 
     # Case 2: Ensure that we can change the underlying table and the materialized view is recreated
     sqlmesh.upsert_model(
@@ -3789,11 +3792,7 @@ def test_materialized_view_evaluation(ctx: TestContext, mocker: MockerFixture):
     with mock.patch.object(logger, "info") as mock_logger:
         sqlmesh.plan(auto_apply=True, no_prompts=True)
 
-        # RisingWave does not need to recreate the mview, all other engines do
-        recreate_view = (
-            "Skipping creation of the view" if dialect == "risingwave" else "Replacing view"
-        )
-        assert any(recreate_view in call[0][0] for call in mock_logger.call_args_list)
+        assert any("Replacing view" in call[0][0] for call in mock_logger.call_args_list)
+
 
-    df = adapter.fetchdf(f"SELECT * FROM {mview_name.sql(dialect=dialect)}")
-    assert df["col"][0] == 2
+    _assert_mview_value(value=2)