* Refactoring

* Making optional but failing requirements mandatory
* Fixing unit tests
* Documentation updates

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

Author: jukiewiczm-marketer

docs/integrations/engines/starrocks.md

@@ -79,7 +79,7 @@ FROM source.user_events;
 
 A `DUPLICATE KEY` table can usually be used as a `FULL` kind model.
 
-### 2) An incremental table (PRIMARY KEY recommended)
+### 2) An incremental table (PRIMARY KEY required)
 
 ```sql
 MODEL (
@@ -129,9 +129,11 @@ MODEL (
 
 ### PRIMARY KEY Type
 
-For incremental models, **PRIMARY KEY tables are needed** (and effectively required for robust deletes) because StarRocks supports *weaker* `DELETE ... WHERE ...` on non-primary-key table types.
+For incremental models, a **PRIMARY KEY table is mandatory**. StarRocks only supports the full `DELETE ... WHERE ...` and `MERGE` semantics that incremental kinds rely on (such as `INCREMENTAL_BY_TIME_RANGE`, `INCREMENTAL_BY_UNIQUE_KEY`, `INCREMENTAL_BY_PARTITION`, and `SCD_TYPE_2`) on PRIMARY KEY tables. On DUPLICATE KEY, UNIQUE KEY, and AGGREGATE KEY tables these operations are not supported well enough.
 
-SQLMesh will apply conservative `WHERE` transformations for compatibility (for example, converting `BETWEEN` to `>= AND <=`, removing boolean literals, and converting `DELETE ... WHERE TRUE` to `TRUNCATE TABLE`). To avoid limitations and keep incremental maintenance reliable, use a `PRIMARY KEY` table by setting `physical_properties.primary_key`.
+SQLMesh enforces this: an incremental model on StarRocks without a primary key fails fast with a clear error. Set `physical_properties.primary_key`, for example `physical_properties (primary_key = (user_id, event_date))`. As a convenience, an `INCREMENTAL_BY_UNIQUE_KEY` model's `unique_key` is automatically promoted to a PRIMARY KEY table.
+
+SQLMesh engine also applies conservative `WHERE` transformations for compatibility (for example, converting `BETWEEN` to `>= AND <=`, removing boolean literals, and converting `DELETE ... WHERE TRUE` to `TRUNCATE TABLE`).
 
 > SQLMesh currently does not support specifying `primary_key` as a model parameter.
 
@@ -160,7 +162,7 @@ GROUP BY user_id, event_date;
 
 ### UNIQUE KEY Type
 
-You can create a UNIQUE KEY table by setting `physical_properties.unique_key`. In most incremental use cases, a PRIMARY KEY table is recommended instead.
+You can create a UNIQUE KEY table by setting `physical_properties.unique_key`. Note that a UNIQUE KEY table is **not** sufficient for incremental models — incremental kinds require a PRIMARY KEY table (see [PRIMARY KEY Type](#primary-key-type)).
 
 **Example:**
 

sqlmesh/core/engine_adapter/base.py

@@ -2728,6 +2728,35 @@ def _build_clustered_by_exp(
     ) -> t.Optional[exp.Cluster]:
         return None
 
+    def adjust_physical_properties_for_incremental(
+        self,
+        physical_properties: t.Dict[str, t.Any],
+        *,
+        requires_delete_capable_table: bool,
+        unique_key: t.Optional[t.List[exp.Expr]],
+        model_name: str,
+    ) -> t.Dict[str, t.Any]:
+        """Adjusts physical properties for an incremental model before the table is created.
+
+        Some engines require a specific physical table layout before they can run the DELETE/MERGE
+        statements that incremental model kinds rely on (e.g. StarRocks only supports those on
+        PRIMARY KEY tables). This hook lets each engine derive or validate the required properties
+        while keeping the generic evaluator free of engine-specific branching.
+
+        Args:
+            physical_properties: The model's physical properties.
+            requires_delete_capable_table: Whether the model kind issues DELETE/MERGE statements
+                (as opposed to append-only INSERTs), as determined by the generic evaluator.
+            unique_key: The model's unique key, populated only when the kind allows promoting it to
+                an engine-specific key (i.e. INCREMENTAL_BY_UNIQUE_KEY); otherwise None.
+            model_name: The model name, for use in diagnostics.
+
+        Returns:
+            The (possibly adjusted) physical properties. Implementations own the given mapping and
+            may mutate it in place; the base implementation returns it unchanged.
+        """
+        return physical_properties
+
     def _build_table_properties_exp(
         self,
         catalog_name: t.Optional[str] = None,

sqlmesh/core/engine_adapter/starrocks.py

@@ -1672,8 +1672,11 @@ class StarRocksEngineAdapter(
     TODO: later, we can add support for INSERT OVERWRITE, even use Primary Key for beter performance
     """
 
-    COMMENT_CREATION_TABLE = CommentCreationTable.IN_SCHEMA_DEF_CTAS
-    """Table comments are added in both CREATE TABLE statement and CTAS"""
+    COMMENT_CREATION_TABLE = CommentCreationTable.IN_SCHEMA_DEF_NO_CTAS
+    """Column comments are added inline in a plain CREATE TABLE, but StarRocks CTAS only accepts a
+    bare column-name list (no types or per-column COMMENT) before AS SELECT. So for CTAS we emit
+    `CREATE TABLE t COMMENT '...' AS SELECT ...` (table comment only) and register column comments
+    afterward via ALTER TABLE ... MODIFY COLUMN ... COMMENT (see _build_create_comment_column_exp)."""
 
     COMMENT_CREATION_VIEW = CommentCreationView.IN_SCHEMA_DEF_NO_COMMANDS
     """View comments are added in CREATE VIEW statement"""
@@ -2105,6 +2108,48 @@ def execute(
             **kwargs,
         )
 
+    def adjust_physical_properties_for_incremental(
+        self,
+        physical_properties: t.Dict[str, t.Any],
+        *,
+        requires_delete_capable_table: bool,
+        unique_key: t.Optional[t.List[exp.Expr]],
+        model_name: str,
+    ) -> t.Dict[str, t.Any]:
+        """Enforce that StarRocks incremental models use a PRIMARY KEY table.
+
+        Incremental kinds rely on DELETE/MERGE statements that StarRocks only supports on PRIMARY
+        KEY tables; DUPLICATE/UNIQUE/AGGREGATE KEY tables reject the predicates SQLMesh generates
+        (e.g. a time-range DELETE with a CAST bound, or any non-key-column predicate). When a
+        unique_key is available (INCREMENTAL_BY_UNIQUE_KEY) we promote it to a PRIMARY KEY;
+        otherwise a PRIMARY KEY must be specified explicitly via physical_properties, and we raise
+        so the failure is clear at creation time rather than producing a broken table.
+
+        The caller owns ``physical_properties`` (it is already a defensive copy), so we mutate and
+        return it in place.
+        """
+        if not requires_delete_capable_table or "primary_key" in physical_properties:
+            return physical_properties
+
+        # Promote the model's unique_key to a PRIMARY KEY table so that complex DELETE/MERGE
+        # statements remain supported.
+        if unique_key:
+            physical_properties["primary_key"] = (
+                unique_key[0] if len(unique_key) == 1 else exp.Tuple(expressions=unique_key)
+            )
+            logger.info(
+                "Model '%s' promoted to PRIMARY KEY table on StarRocks to support rich DELETE operations.",
+                model_name,
+            )
+            return physical_properties
+
+        raise SQLMeshError(
+            f"StarRocks incremental model '{model_name}' requires a PRIMARY KEY table. "
+            "Incremental kinds use DELETE/MERGE operations that StarRocks only supports on PRIMARY KEY "
+            "tables; DUPLICATE/UNIQUE/AGGREGATE KEY tables are not sufficient. "
+            "Specify `physical_properties (primary_key = (...))`, or set `unique_key` on the model."
+        )
+
     # ==================== Table Creation (CORE IMPLEMENTATION) ====================
     def _create_table_from_columns(
         self,
@@ -2546,10 +2591,18 @@ def _build_table_properties_exp(
             properties.append(distributed_prop)
 
         # 5. Handle refresh_property (REFRESH ...)
+        # StarRocks only supports ASYNC materialized views, which require a REFRESH clause.
+        # Synchronous MVs are not supported, so a missing refresh is a hard error rather than
+        # a silent fallback (which would create an undetectable sync MV).
         if is_mv:
             refresh_prop = self._build_refresh_property(table_properties_copy)
-            if refresh_prop:
-                properties.append(refresh_prop)
+            if refresh_prop is None:
+                raise SQLMeshError(
+                    "StarRocks materialized views require a REFRESH clause. "
+                    "Specify at least one of 'refresh_moment' or 'refresh_scheme' in the model's "
+                    "physical_properties (e.g. refresh_scheme = 'ASYNC')."
+                )
+            properties.append(refresh_prop)
 
         # 6. Handle order_by/clustered_by (ORDER BY ...)
         order_prop = self._build_order_by_property(table_properties_copy, clustered_by or None)
@@ -3314,9 +3367,9 @@ def _build_create_comment_table_exp(
         StarRocks uses non-standard syntax for table comments:
             ALTER TABLE {table} COMMENT = '{comment}'
 
-        Note: This method is typically NOT called for StarRocks because:
-        - COMMENT_CREATION_TABLE = IN_SCHEMA_DEF_CTAS
-        - Comments are included directly in CREATE TABLE via SchemaCommentProperty
+        Note: This method is typically NOT called for StarRocks because the table comment is
+        included directly in CREATE TABLE (and CTAS) via SchemaCommentProperty, which StarRocks
+        accepts even for `CREATE TABLE ... COMMENT '...' AS SELECT`.
 
         However, this override is provided for potential future use cases:
         - Modifying comments on existing tables via ALTER TABLE
@@ -3346,15 +3399,13 @@ def _build_create_comment_column_exp(
         """
         Build ALTER TABLE MODIFY COLUMN SQL for column comment modification.
 
-        StarRocks requires column type in MODIFY COLUMN statement:
-            ALTER TABLE {table} MODIFY COLUMN {column} {type} COMMENT '{comment}'
-
-        Note: This method is typically NOT called for StarRocks because:
-        - COMMENT_CREATION_TABLE = IN_SCHEMA_DEF_CTAS
-        - Column comments are included directly in CREATE TABLE DDL
+        StarRocks accepts the comment without re-stating the column type:
+            ALTER TABLE {table} MODIFY COLUMN {column} COMMENT '{comment}'
 
-        However, this override is provided for potential future use cases:
-        - Modifying column comments on existing tables via ALTER TABLE
+        Because COMMENT_CREATION_TABLE = IN_SCHEMA_DEF_NO_CTAS, column comments are inlined for a
+        plain CREATE TABLE but NOT for CTAS (StarRocks rejects types/comments in a CTAS column
+        list). This method is therefore the fallback used to register column comments after a CTAS,
+        and to modify column comments on existing tables.
 
         Args:
             table: Table expression

sqlmesh/core/snapshot/evaluator.py

@@ -2045,38 +2045,36 @@ def run_post_statements(self, snapshot: Snapshot, render_kwargs: t.Any) -> None:
         self.adapter.execute(snapshot.model.render_post_statements(**render_kwargs))
 
 
-def _ensure_primary_key_for_starrocks_when_incremental_by_unique_key(
-    model: Model, physical_properties: t.Optional[t.Dict[str, t.Any]]
+def _adjust_physical_properties_for_engine(
+    adapter: EngineAdapter,
+    model: Model,
+    physical_properties: t.Optional[t.Dict[str, t.Any]],
 ) -> t.Dict[str, t.Any]:
-    """
-    Promote StarRocks incremental-by-unique-key models to PRIMARY KEY tables so that
-    complex DELETE/MERGE statements remain supported.
-    """
+    """Let the target engine adjust/validate physical properties for an incremental model.
 
-    properties = dict(physical_properties or {})
-
-    if (
-        model.dialect != "starrocks"
-        or not model.kind.is_incremental_by_unique_key
-        or "primary_key" in properties
-    ):
-        return properties
-    unique_key: t.Optional[t.List[exp.Expr]] = model.unique_key
-    if unique_key:
-        properties["primary_key"] = (
-            unique_key[0] if len(unique_key) == 1 else exp.Tuple(expressions=unique_key)
-        )
-        logger.info(
-            "Model '%s' promoted to PRIMARY KEY table on StarRocks to support rich DELETE operations.",
-            model.name,
-        )
-    else:
-        logger.warning(
-            f"StarRocks incremental-by-unique-key model '{model.name}' requires a PRIMARY KEY table. "
-            f"Specify `physical_properties['primary_key']` or set `unique_key` on the model.",
-        )
+    The generic responsibility here is to determine, from the model kind, whether the table will
+    be the target of DELETE/MERGE statements (vs. append-only INSERTs) and whether its unique_key
+    may be promoted to an engine-specific key. The engine adapter decides what, if anything, to do
+    with that information (see ``EngineAdapter.adjust_physical_properties_for_incremental``).
+    """
+    kind = model.kind
+
+    # Only incremental kinds that issue DELETE/MERGE need a delete-capable table. Append-only
+    # INCREMENTAL_UNMANAGED (insert_overwrite=False) only does INSERT, so it does not.
+    requires_delete_capable_table = (
+        kind.is_incremental_by_time_range
+        or kind.is_incremental_by_unique_key
+        or kind.is_incremental_by_partition
+        or kind.is_scd_type_2
+        or (isinstance(kind, IncrementalUnmanagedKind) and kind.insert_overwrite)
+    )
 
-    return properties
+    return adapter.adjust_physical_properties_for_incremental(
+        dict(physical_properties or {}),
+        requires_delete_capable_table=requires_delete_capable_table,
+        unique_key=model.unique_key if kind.is_incremental_by_unique_key else None,
+        model_name=model.name,
+    )
 
 
 class MaterializableStrategy(PromotableStrategy, abc.ABC):
@@ -2090,9 +2088,8 @@ def create(
         **kwargs: t.Any,
     ) -> None:
         ctas_query = model.ctas_query(**render_kwargs)
-        physical_properties = kwargs.get("physical_properties", model.physical_properties)
-        physical_properties = _ensure_primary_key_for_starrocks_when_incremental_by_unique_key(
-            model, physical_properties
+        physical_properties = _adjust_physical_properties_for_engine(
+            self.adapter, model, kwargs.get("physical_properties", model.physical_properties)
         )
 
         logger.info("Creating table '%s'", table_name)
@@ -2208,9 +2205,8 @@ def _replace_query_for_model(
             except Exception:
                 columns_to_types, source_columns = None, None
 
-        physical_properties = kwargs.get("physical_properties", model.physical_properties)
-        physical_properties = _ensure_primary_key_for_starrocks_when_incremental_by_unique_key(
-            model, physical_properties
+        physical_properties = _adjust_physical_properties_for_engine(
+            self.adapter, model, kwargs.get("physical_properties", model.physical_properties)
         )
         self.adapter.replace_query(
             name,
@@ -2354,9 +2350,8 @@ def insert(
                 table_name,
                 render_kwargs=render_kwargs,
             )
-            physical_properties = kwargs.get("physical_properties", model.physical_properties)
-            physical_properties = _ensure_primary_key_for_starrocks_when_incremental_by_unique_key(
-                model, physical_properties
+            physical_properties = _adjust_physical_properties_for_engine(
+                self.adapter, model, kwargs.get("physical_properties", model.physical_properties)
             )
             self.adapter.merge(
                 table_name,
@@ -2384,9 +2379,8 @@ def append(
         columns_to_types, source_columns = self._get_target_and_source_columns(
             model, table_name, render_kwargs=render_kwargs
         )
-        physical_properties = kwargs.get("physical_properties", model.physical_properties)
-        physical_properties = _ensure_primary_key_for_starrocks_when_incremental_by_unique_key(
-            model, physical_properties
+        physical_properties = _adjust_physical_properties_for_engine(
+            self.adapter, model, kwargs.get("physical_properties", model.physical_properties)
         )
         self.adapter.merge(
             table_name,
@@ -2594,6 +2588,9 @@ def create(
             columns_to_types = model.columns_to_types_or_raise
             if isinstance(model.kind, SCDType2ByTimeKind):
                 columns_to_types[model.kind.updated_at_name.name] = model.kind.time_data_type
+            physical_properties = _adjust_physical_properties_for_engine(
+                self.adapter, model, kwargs.get("physical_properties", model.physical_properties)
+            )
             self.adapter.create_table(
                 table_name,
                 target_columns_to_types=columns_to_types,
@@ -2602,7 +2599,7 @@ def create(
                 partitioned_by=model.partitioned_by,
                 partition_interval_unit=model.partition_interval_unit,
                 clustered_by=model.clustered_by,
-                table_properties=kwargs.get("physical_properties", model.physical_properties),
+                table_properties=physical_properties,
                 table_description=model.description if is_table_deployable else None,
                 column_descriptions=model.column_descriptions if is_table_deployable else None,
             )
@@ -3179,9 +3176,8 @@ def create(
         if is_table_deployable and is_snapshot_deployable:
             # We could deploy this to prod; create a proper managed table
             logger.info("Creating managed table: %s", table_name)
-            physical_properties = kwargs.get("physical_properties", model.physical_properties)
-            physical_properties = _ensure_primary_key_for_starrocks_when_incremental_by_unique_key(
-                model, physical_properties
+            physical_properties = _adjust_physical_properties_for_engine(
+                self.adapter, model, kwargs.get("physical_properties", model.physical_properties)
             )
             self.adapter.create_managed_table(
                 table_name=table_name,