Docs update + StarRocks AMV with audits gets refreshed immediately

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

Author: jukiewiczm-marketer

docs/integrations/engines/starrocks.md

@@ -485,6 +485,31 @@ FROM user_events
 GROUP BY user_id;
 ```
 
+**Audits on materialized views:**
+
+Audits require data to exist in the materialized view when they run. Because StarRocks refreshes async MVs as background jobs, the data is not guaranteed to be present immediately after the MV is created. To make audits deterministic, when a materialized view has audits SQLMesh issues a synchronous `REFRESH MATERIALIZED VIEW <name> WITH SYNC MODE` right after creating the MV, which blocks until the data is materialized.
+
+For this to work safely, a materialized view with audits **must** set `refresh_moment = 'DEFERRED'`. This prevents StarRocks' automatic (IMMEDIATE) refresh from racing with the synchronous refresh that SQLMesh issues. If the MV has audits and `refresh_moment` is `IMMEDIATE` (or unset, which defaults to `IMMEDIATE` in StarRocks), SQLMesh raises an error before creating the MV.
+
+```sql
+MODEL (
+  name user_summary_mv,
+  kind VIEW (
+    materialized true
+  ),
+  audits (
+    not_null(columns := (user_id))
+  ),
+  physical_properties (
+    -- required when the MV has audits
+    refresh_moment = DEFERRED,
+    refresh_scheme = 'ASYNC'
+  )
+);
+
+SELECT user_id, COUNT(*) AS event_count FROM user_events GROUP BY user_id;
+```
+
 **Other properties:**
 
 You can specify `partitioning`, `distribution`, `order by` and `properties` the same as normal table properties. But notice that only supported MV properties are useful, Refer to StarRocks' doc for MV creation.
@@ -512,6 +537,9 @@ target_columns_to_types = {
 ## Limitations
 
 * **No SYNC MV support**: synchronous materialized views are not supported yet.
+* **`FULL` models are not replaced atomically**: StarRocks does not support `CREATE OR REPLACE TABLE` and has no multi-statement transactions (in version 3.5 and lower), so SQLMesh refreshes a `FULL` model by emptying the existing table (a `TRUNCATE`, or a `DELETE` when a filter applies) and then inserting the new result set as separate, auto-committed statements. There is a brief window between the truncate/delete and the completion of the insert during which the table is empty or partially populated, so readers querying it during that window may see missing or incomplete data. Incremental kinds (e.g. `INCREMENTAL_BY_TIME_RANGE`, `INCREMENTAL_BY_PARTITION`) do not fully eliminate this — StarRocks applies them as the same non-atomic delete-then-insert — but they narrow the affected rows to the partition/time range being processed rather than emptying the whole table, so unaffected partitions remain readable throughout. SQLMesh has no way to make these replacements atomic on StarRocks 3.5 and lower.
+
+    Future work: this PR targeted StarRocks 3.5, but StarRocks has since expanded its capabilities considerably (the integration now runs against 4.1). Later work should investigate using `INSERT OVERWRITE` together with the transactional/atomic-swap guarantees available in newer StarRocks versions to close this gap (see the `INSERT_OVERWRITE_STRATEGY` and `SUPPORTS_TRANSACTIONS` flags in the StarRocks engine adapter).
 * **No tuple IN**: StarRocks does not support `(c1, c2) IN ((v1, v2), ...)`.
 * **No `SELECT ... FOR UPDATE`**: StarRocks is an OLAP database and does not support row locks; SQLMesh removes `FOR UPDATE` when executing SQLGlot expressions.
 * **RENAME caveat**: `ALTER TABLE db.old RENAME db.new` is not supported; the `RENAME` target cannot be qualified with a database name.

sqlmesh/core/engine_adapter/starrocks.py

@@ -2310,6 +2310,13 @@ def create_view(
             )
 
         # MATERIALIZED VIEW path
+        # MVs with audits get a synchronous refresh after creation (see _create_materialized_view),
+        # which requires REFRESH DEFERRED. Validate before the drop so we fail without destroying
+        # an existing MV.
+        has_audits = bool((materialized_properties or {}).get("has_audits"))
+        if has_audits:
+            self._validate_deferred_refresh_for_audits(view_name, view_properties)
+
         if replace:
             # Avoid DROP MATERIALIZED VIEW failure when an object with the same name exists but is not an MV.
             self.drop_data_object_on_type_mismatch(
@@ -2431,6 +2438,16 @@ def _create_materialized_view(
                 quote_identifiers=self.QUOTE_IDENTIFIERS_IN_VIEWS,
             )
 
+        # MVs with audits are created with REFRESH DEFERRED (enforced in create_view), so StarRocks
+        # does not populate them on creation. Audits need data, so block on a synchronous refresh.
+        if bool((materialized_properties or {}).get("has_audits")):
+            refresh_sql = (
+                f"REFRESH MATERIALIZED VIEW "
+                f"{exp.to_table(view_name).sql(dialect=self.dialect, identify=True)} "
+                f"WITH SYNC MODE"
+            )
+            self.execute(refresh_sql)
+
         self._clear_data_object_cache(view_name)
 
     def _build_materialized_view_schema_exp(
@@ -3020,6 +3037,37 @@ def _build_distributed_by_property(
         )
         return result
 
+    def _validate_deferred_refresh_for_audits(
+        self,
+        view_name: TableName,
+        view_properties: t.Optional[t.Dict[str, exp.Expr]],
+    ) -> None:
+        """
+        Ensure a materialized view with audits uses REFRESH DEFERRED.
+
+        StarRocks audits require data to exist in the MV, so SQLMesh issues an explicit synchronous
+        `REFRESH MATERIALIZED VIEW ... WITH SYNC MODE` right after creating the MV. For that to be
+        deterministic, the MV must use `refresh_moment = 'DEFERRED'`; otherwise StarRocks' automatic
+        (IMMEDIATE) refresh would run concurrently and race with the explicit one. A missing
+        refresh_moment defaults to IMMEDIATE in StarRocks, so it is rejected as well.
+        """
+        refresh_moment = (view_properties or {}).get("refresh_moment")
+        normalized = (
+            PropertyValidator.validate_and_normalize_property("refresh_moment", refresh_moment)
+            if refresh_moment is not None
+            else None
+        )
+        if normalized != "DEFERRED":
+            raise SQLMeshError(
+                f"[StarRocks] Materialized view '{exp.to_table(view_name).sql(dialect=self.dialect)}' "
+                "has audits, which require a synchronous refresh after creation. This is only "
+                "supported with deferred refresh, so the model must set "
+                "`refresh_moment = 'DEFERRED'` in its physical_properties "
+                f"(got {normalized or 'no refresh_moment; StarRocks defaults to IMMEDIATE'}). "
+                "DEFERRED prevents StarRocks' "
+                "automatic refresh from racing with the synchronous refresh SQLMesh issues."
+            )
+
     def _build_refresh_property(
         self, table_properties: t.Dict[str, t.Any]
     ) -> t.Optional[exp.RefreshTriggerProperty]:

sqlmesh/core/snapshot/evaluator.py

@@ -2753,13 +2753,12 @@ def insert(
 
         logger.info("Replacing view '%s'", table_name)
         materialized_properties = None
-        if is_materialized_view and (
-            model.partitioned_by or model.partition_interval_unit or model.clustered_by
-        ):
+        if is_materialized_view:
             materialized_properties = {
                 "partitioned_by": model.partitioned_by,
                 "partition_interval_unit": model.partition_interval_unit,
                 "clustered_by": model.clustered_by,
+                "has_audits": bool(model.audits_with_args),
             }
         self.adapter.create_view(
             table_name,
@@ -2818,6 +2817,7 @@ def create(
                 "partitioned_by": model.partitioned_by,
                 "clustered_by": model.clustered_by,
                 "partition_interval_unit": model.partition_interval_unit,
+                "has_audits": bool(model.audits_with_args),
             }
         self.adapter.create_view(
             table_name,
@@ -2853,11 +2853,22 @@ def migrate(
             execution_time=now(), snapshots=kwargs["snapshots"], engine_adapter=self.adapter
         )
 
+        is_materialized_view = self._is_materialized_view(model)
+        materialized_properties = None
+        if is_materialized_view:
+            materialized_properties = {
+                "partitioned_by": model.partitioned_by,
+                "clustered_by": model.clustered_by,
+                "partition_interval_unit": model.partition_interval_unit,
+                "has_audits": bool(model.audits_with_args),
+            }
+
         self.adapter.create_view(
             target_table_name,
             model.render_query_or_raise(**render_kwargs),
             model.columns_to_types,
-            materialized=self._is_materialized_view(model),
+            materialized=is_materialized_view,
+            materialized_properties=materialized_properties,
             view_properties=model.render_physical_properties(**render_kwargs),
             table_description=model.description,
             column_descriptions=model.column_descriptions,

tests/core/engine_adapter/test_starrocks.py

@@ -350,6 +350,118 @@ def test_create_materialized_view_without_refresh_raises(
                 view_properties={"replication_num": exp.Literal.string("1")},
             )
 
+    def test_create_materialized_view_with_audits_emits_sync_refresh(
+        self, make_mocked_engine_adapter: t.Callable[..., StarRocksEngineAdapter]
+    ):
+        """When an MV has audits, SQLMesh must synchronously refresh it right after creation.
+
+        Audits require data to exist in the MV. With REFRESH DEFERRED, StarRocks does not populate
+        the MV on creation, so SQLMesh issues an explicit `REFRESH MATERIALIZED VIEW ... WITH SYNC
+        MODE` that blocks until the data is materialized.
+        """
+        adapter = make_mocked_engine_adapter(StarRocksEngineAdapter)
+        adapter.create_view(
+            "test_mv",
+            parse_one("SELECT a FROM tbl"),
+            materialized=True,
+            target_columns_to_types={"a": exp.DataType.build("INT")},
+            materialized_properties={"has_audits": True},
+            view_properties={
+                "refresh_moment": exp.Var(this="DEFERRED"),
+                "refresh_scheme": exp.Var(this="ASYNC"),
+            },
+        )
+
+        calls = to_sql_calls(adapter)
+        assert calls[0] == "DROP MATERIALIZED VIEW IF EXISTS `test_mv`"
+        assert "CREATE MATERIALIZED VIEW" in calls[1]
+        assert "REFRESH DEFERRED" in calls[1]
+        assert calls[2] == "REFRESH MATERIALIZED VIEW `test_mv` WITH SYNC MODE"
+
+    def test_create_materialized_view_with_audits_emits_sync_refresh_on_first_create(
+        self, make_mocked_engine_adapter: t.Callable[..., StarRocksEngineAdapter]
+    ):
+        """The sync refresh must also fire on first-time creation (replace=False).
+
+        ViewStrategy.create calls create_view with replace=False, so the DROP is skipped, but the
+        synchronous refresh still needs to populate the MV before audits run.
+        """
+        adapter = make_mocked_engine_adapter(StarRocksEngineAdapter)
+        adapter.create_view(
+            "test_mv",
+            parse_one("SELECT a FROM tbl"),
+            replace=False,
+            materialized=True,
+            target_columns_to_types={"a": exp.DataType.build("INT")},
+            materialized_properties={"has_audits": True},
+            view_properties={
+                "refresh_moment": exp.Var(this="DEFERRED"),
+                "refresh_scheme": exp.Var(this="ASYNC"),
+            },
+        )
+
+        calls = to_sql_calls(adapter)
+        assert all("DROP MATERIALIZED VIEW" not in sql for sql in calls)
+        assert "CREATE MATERIALIZED VIEW" in calls[0]
+        assert "REFRESH DEFERRED" in calls[0]
+        assert calls[1] == "REFRESH MATERIALIZED VIEW `test_mv` WITH SYNC MODE"
+
+    def test_create_materialized_view_with_audits_immediate_refresh_raises(
+        self, make_mocked_engine_adapter: t.Callable[..., StarRocksEngineAdapter]
+    ):
+        """An MV with audits must use REFRESH DEFERRED; IMMEDIATE must raise and not create anything."""
+        adapter = make_mocked_engine_adapter(StarRocksEngineAdapter)
+        with pytest.raises(SQLMeshError, match="DEFERRED"):
+            adapter.create_view(
+                "test_mv",
+                parse_one("SELECT a FROM tbl"),
+                materialized=True,
+                target_columns_to_types={"a": exp.DataType.build("INT")},
+                materialized_properties={"has_audits": True},
+                view_properties={
+                    "refresh_moment": exp.Var(this="IMMEDIATE"),
+                    "refresh_scheme": exp.Var(this="ASYNC"),
+                },
+            )
+
+        # Fail-fast: nothing should have been dropped or created.
+        assert all("CREATE MATERIALIZED VIEW" not in sql for sql in to_sql_calls(adapter))
+        assert all("DROP MATERIALIZED VIEW" not in sql for sql in to_sql_calls(adapter))
+
+    def test_create_materialized_view_with_audits_missing_refresh_moment_raises(
+        self, make_mocked_engine_adapter: t.Callable[..., StarRocksEngineAdapter]
+    ):
+        """A missing refresh_moment defaults to IMMEDIATE in StarRocks, so audits must raise."""
+        adapter = make_mocked_engine_adapter(StarRocksEngineAdapter)
+        with pytest.raises(SQLMeshError, match="DEFERRED"):
+            adapter.create_view(
+                "test_mv",
+                parse_one("SELECT a FROM tbl"),
+                materialized=True,
+                target_columns_to_types={"a": exp.DataType.build("INT")},
+                materialized_properties={"has_audits": True},
+                view_properties={"refresh_scheme": exp.Var(this="ASYNC")},
+            )
+
+    def test_create_materialized_view_without_audits_does_not_sync_refresh(
+        self, make_mocked_engine_adapter: t.Callable[..., StarRocksEngineAdapter]
+    ):
+        """Without audits, no synchronous refresh is issued and IMMEDIATE refresh is allowed."""
+        adapter = make_mocked_engine_adapter(StarRocksEngineAdapter)
+        adapter.create_view(
+            "test_mv",
+            parse_one("SELECT a FROM tbl"),
+            materialized=True,
+            target_columns_to_types={"a": exp.DataType.build("INT")},
+            materialized_properties={"has_audits": False},
+            view_properties={
+                "refresh_moment": exp.Var(this="IMMEDIATE"),
+                "refresh_scheme": exp.Var(this="ASYNC"),
+            },
+        )
+
+        assert all("WITH SYNC MODE" not in sql for sql in to_sql_calls(adapter))
+
     def test_does_not_recreate_materialized_view_on_evaluation(self):
         """StarRocks async MVs maintain themselves, so SQLMesh must not recreate them on every run.
 

tests/core/test_snapshot_evaluator.py

@@ -843,7 +843,12 @@ def test_evaluate_materialized_view_with_execution_time_macro(
         view_properties={},
         table_description=None,
         column_descriptions={},
-        materialized_properties=None,
+        materialized_properties={
+            "partitioned_by": [],
+            "partition_interval_unit": None,
+            "clustered_by": [],
+            "has_audits": False,
+        },
     )
 
 
@@ -1305,6 +1310,7 @@ def test_create_materialized_view(mocker: MockerFixture, adapter_mock, make_snap
             "clustered_by": [],
             "partition_interval_unit": None,
             "partitioned_by": [],
+            "has_audits": False,
         },
         view_properties={},
         table_description=None,
@@ -1354,6 +1360,7 @@ def test_create_view_with_properties(mocker: MockerFixture, adapter_mock, make_s
             "clustered_by": [],
             "partition_interval_unit": None,
             "partitioned_by": [],
+            "has_audits": False,
         },
         table_description=None,
         replace=False,
@@ -1364,6 +1371,45 @@ def test_create_view_with_properties(mocker: MockerFixture, adapter_mock, make_s
     )
 
 
+def test_create_materialized_view_with_audits_sets_has_audits(
+    mocker: MockerFixture, adapter_mock, make_snapshot
+):
+    """A materialized view model with audits must propagate has_audits=True to the adapter.
+
+    Engines like StarRocks rely on this flag to synchronously refresh the MV before audits run.
+    """
+    adapter_mock.get_data_objects.return_value = []
+    adapter_mock.table_exists.return_value = False
+
+    evaluator = SnapshotEvaluator(adapter_mock)
+
+    model = load_sql_based_model(
+        parse(  # type: ignore
+            """
+            MODEL (
+                name test_schema.test_model,
+                kind VIEW (
+                    materialized true
+                ),
+                audits (
+                    not_null(columns := (a))
+                )
+            );
+
+            SELECT a::int FROM tbl;
+            """
+        ),
+    )
+
+    snapshot = make_snapshot(model)
+    snapshot.categorize_as(SnapshotChangeCategory.BREAKING)
+
+    evaluator.create([snapshot], {})
+
+    _, kwargs = adapter_mock.create_view.call_args
+    assert kwargs["materialized_properties"]["has_audits"] is True
+
+
 def test_promote_model_info(mocker: MockerFixture, make_snapshot):
     adapter_mock = mocker.patch("sqlmesh.core.engine_adapter.EngineAdapter")
     adapter_mock.dialect = "duckdb"