Make StarRocks AMVs not recreate themselves during sqlmesh run

Signed-off-by: Mateusz Jukiewicz <mateusz@marketer.tech>

Author: jukiewiczm-marketer

docs/integrations/engines/starrocks.md

@@ -491,7 +491,11 @@ You can specify `partitioning`, `distribution`, `order by` and `properties` the
 
 **Notes:**
 
-* If you create materialized views with `replace=true`, SQLMesh may drop and recreate the MV. When an MV is dropped, its data is removed and the MV must be refreshed/rebuilt again.
+* SQLMesh does not recreate materialized views on every `sqlmesh run`. Once an MV exists, SQLMesh leaves it in place and lets StarRocks keep it current. This is intentional:
+    * StarRocks async MVs revalidate themselves automatically, even when the underlying data is dropped, so a periodic drop-and-recreate is unnecessary.
+    * StarRocks async MVs either refresh automatically (per their `refresh_scheme`) or can be refreshed explicitly with `REFRESH MATERIALIZED VIEW`, which also enables partition-level (incremental) refresh. A SQLMesh-driven recreate would instead force a full rebuild.
+
+    The MV is (re)built only when it does not yet exist — for example when you first deploy it, or when a model change produces a new version. To change a materialized view's definition, update the model and run `sqlmesh plan`.
 * There are some restriction for `partitioning`, you need to refer StarRocks' doc for MV partitioning specification.
 * StarRocks MV schema supports a column list but does **not** support explicit data types in that list. Column data types come from the `AS SELECT ...` query.
 * If you create MVs from a dataframe via the Python API, provide `target_columns_to_types` (a `Dict[str, exp.DataType]`). If you don't care about exact types, you can set all columns to `VARCHAR` as a fallback:

sqlmesh/core/engine_adapter/base.py

@@ -114,6 +114,7 @@ class EngineAdapter:
     SCHEMA_DIFFER_KWARGS: t.Dict[str, t.Any] = {}
     SUPPORTS_TUPLE_IN = True
     HAS_VIEW_BINDING = False
+    RECREATE_MATERIALIZED_VIEW_ON_EVALUATION = True
     SUPPORTS_REPLACE_TABLE = True
     SUPPORTS_GRANTS = False
     DEFAULT_CATALOG_TYPE = DIALECT

sqlmesh/core/engine_adapter/starrocks.py

@@ -1691,6 +1691,13 @@ class StarRocksEngineAdapter(
     implement custom MV schema rendering in create_view/_create_materialized_view.
     """
 
+    RECREATE_MATERIALIZED_VIEW_ON_EVALUATION = False
+    """
+    StarRocks async materialized views maintain themselves: they revalidate automatically even if the
+    underlying data is dropped, and the data is kept current either by StarRocks' automatic refresh or
+    by an explicit `REFRESH MATERIALIZED VIEW` (which also enables partition-level incremental refresh).
+    """
+
     SUPPORTS_REPLACE_TABLE = False
     """No REPLACE TABLE syntax; use DROP + CREATE instead"""
 

sqlmesh/core/snapshot/evaluator.py

@@ -2734,6 +2734,19 @@ def insert(
             is_materialized_view and is_first_insert
         )
 
+        # Some engines (e.g. StarRocks) maintain materialized views automatically (via auto/scheduled
+        # REFRESH) and can only recreate them via a destructive DROP + CREATE, which deletes the
+        # materialized data and forces a full rebuild. For those, an existing MV must not be recreated
+        # on routine evaluation (e.g. every `sqlmesh run`); only build it on the first insert (a new
+        # version) or when a rebuild is explicitly forced (intervals cleared by `should_force_rebuild`,
+        # which sets `is_first_insert`).
+        if (
+            is_materialized_view
+            and not is_first_insert
+            and not self.adapter.RECREATE_MATERIALIZED_VIEW_ON_EVALUATION
+        ):
+            must_recreate_view = False
+
         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/test_starrocks.py

@@ -350,6 +350,15 @@ def test_create_materialized_view_without_refresh_raises(
                 view_properties={"replication_num": exp.Literal.string("1")},
             )
 
+    def test_does_not_recreate_materialized_view_on_evaluation(self):
+        """StarRocks async MVs maintain themselves, so SQLMesh must not recreate them on every run.
+
+        The adapter opts out of per-evaluation recreation via
+        RECREATE_MATERIALIZED_VIEW_ON_EVALUATION = False, which the evaluator's ViewStrategy honors
+        for materialized views that already exist.
+        """
+        assert StarRocksEngineAdapter.RECREATE_MATERIALIZED_VIEW_ON_EVALUATION is False
+
     def test_delete_where_true_optimization(
         self, make_mocked_engine_adapter: t.Callable[..., StarRocksEngineAdapter]
     ):

tests/core/test_snapshot_evaluator.py

@@ -707,6 +707,61 @@ def test_evaluate_materialized_view(
     assert adapter_mock.create_view.call_count == 1
 
 
+@pytest.mark.parametrize(
+    "view_exists, has_intervals, expected_create_view_calls",
+    [
+        # MV already exists and has intervals -> routine evaluation (e.g. `sqlmesh run`): do NOT recreate
+        (True, True, 0),
+        # MV does not exist yet -> first build: create it
+        (False, False, 1),
+        # MV missing but intervals present -> still a first insert (table gone): rebuild it
+        (False, True, 1),
+    ],
+)
+def test_evaluate_materialized_view_not_recreated_on_evaluation(
+    adapter_mock,
+    make_snapshot,
+    view_exists: bool,
+    has_intervals: bool,
+    expected_create_view_calls: int,
+):
+    # Engines that maintain MVs themselves (e.g. StarRocks) opt out of recreating an existing MV on
+    # every evaluation by setting RECREATE_MATERIALIZED_VIEW_ON_EVALUATION = False.
+    adapter_mock.RECREATE_MATERIALIZED_VIEW_ON_EVALUATION = False
+    adapter_mock.table_exists.return_value = view_exists
+    evaluator = SnapshotEvaluator(adapter_mock)
+
+    model = load_sql_based_model(
+        parse(  # type: ignore
+            """
+            MODEL (
+                name test_schema.test_model,
+                kind VIEW (
+                    materialized true
+                )
+            );
+
+            SELECT a::int FROM tbl;
+            """
+        ),
+    )
+
+    snapshot = make_snapshot(model)
+    snapshot.categorize_as(SnapshotChangeCategory.BREAKING)
+    if has_intervals:
+        snapshot.add_interval("2023-01-01", "2023-01-01")
+
+    evaluator.evaluate(
+        snapshot,
+        start="2020-01-01",
+        end="2020-01-02",
+        execution_time="2020-01-02",
+        snapshots={},
+    )
+
+    assert adapter_mock.create_view.call_count == expected_create_view_calls
+
+
 def test_evaluate_materialized_view_with_partitioned_by_cluster_by(
     mocker: MockerFixture, adapter_mock, make_snapshot
 ):