Fix(databricks): Add column type only along materialized view comments

sqlmesh/core/engine_adapter/base.py

@@ -811,6 +811,7 @@ def _build_schema_exp(
         column_descriptions: t.Optional[t.Dict[str, str]] = None,
         expressions: t.Optional[t.List[exp.PrimaryKey]] = None,
         is_view: bool = False,
+        materialized: bool = False,
     ) -> exp.Schema:
         """
         Build a schema expression for a table, columns, column comments, and additional schema properties.
@@ -823,6 +824,7 @@ def _build_schema_exp(
                 target_columns_to_types=target_columns_to_types,
                 column_descriptions=column_descriptions,
                 is_view=is_view,
+                materialized=materialized,
             )
             + expressions,
         )
@@ -832,6 +834,7 @@ def _build_column_defs(
         target_columns_to_types: t.Dict[str, exp.DataType],
         column_descriptions: t.Optional[t.Dict[str, str]] = None,
         is_view: bool = False,
+        materialized: bool = False,
     ) -> t.List[exp.ColumnDef]:
         engine_supports_schema_comments = (
             self.COMMENT_CREATION_VIEW.supports_schema_def
@@ -1260,7 +1263,11 @@ def create_view(
         schema: t.Union[exp.Table, exp.Schema] = exp.to_table(view_name)
         if target_columns_to_types:
             schema = self._build_schema_exp(
-                exp.to_table(view_name), target_columns_to_types, column_descriptions, is_view=True
+                exp.to_table(view_name),
+                target_columns_to_types,
+                column_descriptions,
+                is_view=True,
+                materialized=materialized,
             )
 
         properties = create_kwargs.pop("properties", None)

sqlmesh/core/engine_adapter/databricks.py

@@ -400,11 +400,14 @@ def _build_column_defs(
         target_columns_to_types: t.Dict[str, exp.DataType],
         column_descriptions: t.Optional[t.Dict[str, str]] = None,
         is_view: bool = False,
+        materialized: bool = False,
     ) -> t.List[exp.ColumnDef]:
         # Databricks requires column types to be specified when adding column comments
         # in CREATE MATERIALIZED VIEW statements. Override is_view to False to force
         # column types to be included when comments are present.
-        if is_view and column_descriptions:
+        if is_view and materialized and column_descriptions:
             is_view = False
 
-        return super()._build_column_defs(target_columns_to_types, column_descriptions, is_view)
+        return super()._build_column_defs(
+            target_columns_to_types, column_descriptions, is_view, materialized
+        )

sqlmesh/core/engine_adapter/trino.py

@@ -284,6 +284,7 @@ def _build_schema_exp(
         column_descriptions: t.Optional[t.Dict[str, str]] = None,
         expressions: t.Optional[t.List[exp.PrimaryKey]] = None,
         is_view: bool = False,
+        materialized: bool = False,
     ) -> exp.Schema:
         if "delta_lake" in self.get_catalog_type_from_table(table):
             target_columns_to_types = self._to_delta_ts(target_columns_to_types)

tests/core/engine_adapter/test_databricks.py

@@ -406,6 +406,36 @@ def test_materialized_view_with_column_comments(
     ]
 
 
+def test_regular_view_with_column_comments(
+    mocker: MockFixture, make_mocked_engine_adapter: t.Callable
+):
+    mocker.patch(
+        "sqlmesh.core.engine_adapter.databricks.DatabricksEngineAdapter.set_current_catalog"
+    )
+    adapter = make_mocked_engine_adapter(DatabricksEngineAdapter, default_catalog="test_catalog")
+    mocker.patch.object(adapter, "get_current_catalog", return_value="test_catalog")
+
+    adapter.create_view(
+        "test_view",
+        parse_one("SELECT a, b FROM source_table"),
+        target_columns_to_types={
+            "a": exp.DataType.build("INT"),
+            "b": exp.DataType.build("STRING"),
+        },
+        materialized=False,
+        column_descriptions={
+            "a": "column a description",
+            "b": "column b description",
+        },
+    )
+
+    sql_calls = to_sql_calls(adapter)
+    # Regular views should NOT include column types even when column comments are present
+    assert sql_calls == [
+        "CREATE OR REPLACE VIEW `test_view` (`a` COMMENT 'column a description', `b` COMMENT 'column b description') AS SELECT `a`, `b` FROM `source_table`",
+    ]
+
+
 def test_create_table_clustered_by(mocker: MockFixture, make_mocked_engine_adapter: t.Callable):
     mocker.patch(
         "sqlmesh.core.engine_adapter.databricks.DatabricksEngineAdapter.set_current_catalog"