Fix: Include column types in Databricks materialized views with comments

sqlmesh/core/engine_adapter/databricks.py

@@ -394,3 +394,19 @@ def _build_table_properties_exp(
             expressions.append(clustered_by_exp)
             properties = exp.Properties(expressions=expressions)
         return properties
+
+    def _build_column_defs(
+        self,
+        target_columns_to_types: t.Dict[str, exp.DataType],
+        column_descriptions: t.Optional[t.Dict[str, str]] = None,
+        is_view: 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:
+            engine_supports_schema_comments = self.COMMENT_CREATION_VIEW.supports_schema_def
+            if engine_supports_schema_comments and self.comments_enabled:
+                is_view = False
+
+        return super()._build_column_defs(target_columns_to_types, column_descriptions, is_view)

tests/core/engine_adapter/test_databricks.py

@@ -376,6 +376,36 @@ def test_materialized_view_properties(mocker: MockFixture, make_mocked_engine_ad
     ]
 
 
+def test_materialized_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=True,
+        column_descriptions={
+            "a": "column a description",
+            "b": "column b description",
+        },
+    )
+
+    sql_calls = to_sql_calls(adapter)
+    # Databricks requires column types when column comments are present in materialized views
+    assert sql_calls == [
+        "CREATE OR REPLACE MATERIALIZED VIEW `test_view` (`a` INT COMMENT 'column a description', `b` STRING 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"