Don't recreate mview for RisingWave

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,13 +2282,12 @@ def insert(
         render_kwargs: t.Dict[str, t.Any],
         **kwargs: t.Any,
     ) -> None:
-        snapshot = kwargs["snapshot"]
-
-        if (
-            not snapshot.is_materialized_view
-            and self.adapter.HAS_VIEW_BINDING
-            and self.adapter.table_exists(table_name)
-        ):
+        # 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
+        if self.adapter.table_exists(table_name) and not must_recreate_view:
             logger.info("Skipping creation of the view '%s'", table_name)
             return
 
@@ -2297,7 +2296,7 @@ def insert(
             table_name,
             query_or_df,
             model.columns_to_types,
-            replace=snapshot.is_materialized_view or not self.adapter.HAS_VIEW_BINDING,
+            replace=must_recreate_view,
             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

@@ -8,11 +8,13 @@
 import typing as t
 import shutil
 from datetime import datetime, timedelta
+import logging
 
 import numpy as np  # noqa: TID253
 import pandas as pd  # noqa: TID253
 import pytest
 import pytz
+from unittest import mock
 from sqlglot import exp, parse_one
 from sqlglot.optimizer.normalize_identifiers import normalize_identifiers
 from sqlglot.optimizer.qualify_columns import quote_identifiers
@@ -3733,7 +3735,7 @@ def _set_config(_gateway: str, config: Config) -> None:
     ]
 
 
-def test_materialized_view_evaluation(ctx: TestContext):
+def test_materialized_view_evaluation(ctx: TestContext, mocker: MockerFixture):
     adapter = ctx.engine_adapter
     dialect = ctx.dialect
     if not adapter.SUPPORTS_MATERIALIZED_VIEWS:
@@ -3784,7 +3786,16 @@ def test_materialized_view_evaluation(ctx: TestContext):
         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)
+    logger = logging.getLogger("sqlmesh.core.snapshot.evaluator")
+
+    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)
 
     df = adapter.fetchdf(f"SELECT * FROM {mview_name.sql(dialect=dialect)}")
     assert df["col"][0] == 2