Set column comments on create for MV/ST (#1046)

Author: ericj-db

CHANGELOG.md

@@ -6,6 +6,7 @@
 - Fix bug where incorrect insert overwrite exception was being thrown when overriding compute at model level ([1032](https://github.com/databricks/dbt-databricks/issues/1032))
 - Fix bug that was causing failures when multiple foreign key constraints are defined in a model ([1034](https://github.com/databricks/dbt-databricks/issues/1034))
 - Fix location_root config not being treated case sensitive ([1037](https://github.com/databricks/dbt-databricks/issues/1037))
+- Fix column comments for streaming tables and materialized views ([1049](https://github.com/databricks/dbt-databricks/issues/1049))
 
 ## dbt-databricks 1.10.2 (May 21, 2025)
 

dbt/include/databricks/macros/materializations/materialized_view.sql

@@ -70,8 +70,6 @@
     {% set should_revoke = should_revoke(existing_relation, full_refresh_mode=True) %}
     {% do apply_grants(target_relation, grant_config, should_revoke=should_revoke) %}
 
-    {% do persist_docs(target_relation, model, for_relation=False) %}
-
     {{ run_hooks(post_hooks, inside_transaction=True) }}
 
 {% endmacro %}

dbt/include/databricks/macros/materializations/streaming_table.sql

@@ -69,8 +69,6 @@
     {% set should_revoke = should_revoke(existing_relation, full_refresh_mode=True) %}
     {% do apply_grants(target_relation, grant_config, should_revoke=should_revoke) %}
 
-    {% do persist_docs(target_relation, model, for_relation=False) %}
-
     {{ run_hooks(post_hooks, inside_transaction=True) }}
 
 {% endmacro %}

dbt/include/databricks/macros/relations/materialized_view/create.sql

@@ -4,7 +4,15 @@
   {%- set tblproperties = materialized_view.config["tblproperties"].tblproperties -%}
   {%- set comment = materialized_view.config["comment"].comment -%}
   {%- set refresh = materialized_view.config["refresh"] -%}
-  create materialized view {{ relation.render() }}
+
+  {%- set columns = adapter.get_column_schema_from_query(sql) -%}
+  {%- set model_columns = model.get('columns', {}) -%}
+  {%- set model_constraints = model.get('constraints', []) -%}
+  {%- set columns_and_constraints = adapter.parse_columns_and_constraints(columns, model_columns, model_constraints) -%}
+  {%- set target_relation = relation.enrich(columns_and_constraints[1]) -%}
+
+  create materialized view {{ target_relation.render() }}
+    {{ get_column_and_constraints_sql(target_relation, columns_and_constraints[0]) }}
     {{ get_create_sql_partition_by(partition_by) }}
     {{ get_create_sql_comment(comment) }}
     {{ get_create_sql_tblproperties(tblproperties) }}

dbt/include/databricks/macros/relations/streaming_table/create.sql

@@ -9,7 +9,14 @@
   {%- set comment = streaming_table.config["comment"].comment -%}
   {%- set refresh = streaming_table.config["refresh"] -%}
 
+  {%- set analysis_sql = sql | replace('STREAM ', '') | replace('stream ', '') -%}
+  {%- set columns = adapter.get_column_schema_from_query(analysis_sql) -%}
+  {%- set model_columns = model.get('columns', {}) -%}
+  {%- set columns_and_constraints = adapter.parse_columns_and_constraints(columns, model_columns, []) -%}
+
+  {#-- We don't enrich the relation with model constraints because they are not supported for streaming tables --#}
   CREATE STREAMING TABLE {{ relation.render() }}
+    {{ get_column_and_constraints_sql(relation, columns_and_constraints[0]) }}
     {{ get_create_sql_partition_by(partition_by) }}
     {{ get_create_sql_comment(comment) }}
     {{ get_create_sql_tblproperties(tblproperties) }}

tests/functional/adapter/materialized_view_tests/fixtures.py

@@ -44,3 +44,23 @@ def query_relation_type(project, relation: BaseRelation) -> Optional[str]:
 ) }}
 select * from {{ ref('my_seed') }}
 """
+
+materialized_view_schema = """
+version: 2
+
+models:
+  - name: my_materialized_view
+    columns:
+      - name: id
+        description: "The unique identifier for each record"
+        constraints:
+          - type: not_null
+      - name: value
+    constraints:
+        - type: primary_key
+          columns: [id]
+    config:
+      persist_docs:
+        relation: true
+        columns: true
+"""

tests/functional/adapter/materialized_view_tests/test_basic.py

@@ -4,6 +4,7 @@
 
 from dbt.adapters.base.relation import BaseRelation
 from dbt.tests import util
+from dbt.tests.adapter.materialized_view import files
 from dbt.tests.adapter.materialized_view.basic import MaterializedViewBasic
 from tests.functional.adapter.materialized_view_tests import fixtures
 
@@ -29,6 +30,15 @@ def query_relation_type(project, relation: BaseRelation) -> Optional[str]:
 @pytest.mark.dlt
 @pytest.mark.skip_profile("databricks_cluster", "databricks_uc_cluster")
 class TestMaterializedViews(TestMaterializedViewsMixin, MaterializedViewBasic):
+    @pytest.fixture(scope="class", autouse=True)
+    def models(self):
+        yield {
+            "my_table.sql": files.MY_TABLE,
+            "my_view.sql": files.MY_VIEW,
+            "my_materialized_view.sql": files.MY_MATERIALIZED_VIEW,
+            "schema.yml": fixtures.materialized_view_schema,
+        }
+
     def test_table_replaces_materialized_view(self, project, my_materialized_view):
         util.run_dbt(["run", "--models", my_materialized_view.identifier])
         assert self.query_relation_type(project, my_materialized_view) == "materialized_view"
@@ -70,3 +80,42 @@ def test_materialized_view_replaces_view(self, project, my_view):
         util.run_dbt(["run", "--models", my_view.identifier])
         # UC doesn't sync metadata fast enough for this to pass consistently
         # assert self.query_relation_type(project, my_view) == "materialized_view"
+
+    def test_create_materialized_view_with_comment_and_constraints(
+        self, project, my_materialized_view
+    ):
+        util.run_dbt(["run", "--models", my_materialized_view.identifier])
+        assert self.query_relation_type(project, my_materialized_view) == "materialized_view"
+
+        # verify the non-null constraint and column comment are persisted on create
+        results = project.run_sql(
+            f"""
+            SELECT
+                is_nullable,
+                comment
+            FROM {project.database}.information_schema.columns
+            WHERE table_catalog = '{project.database}'
+              AND table_schema = '{project.test_schema}'
+              AND table_name = '{my_materialized_view.identifier}'
+              AND column_name = 'id'""",
+            fetch="all",
+        )
+        row = results[0]
+        assert row[0] == "NO"
+        assert row[1] == "The unique identifier for each record"
+        # Verify primary key constraint is persisted
+        results = project.run_sql(
+            f"""
+            SELECT
+                constraint_name,
+                column_name
+            FROM {project.database}.information_schema.key_column_usage
+            WHERE table_catalog = '{project.database}'
+              AND table_schema = '{project.test_schema}'
+              AND table_name = '{my_materialized_view.identifier}'
+            """,
+            fetch="all",
+        )
+        assert len(results) == 1
+        assert results[0][0] == "my_materialized_view_pk"
+        assert results[0][1] == "id"

tests/functional/adapter/streaming_tables/fixtures.py

@@ -51,3 +51,31 @@ def query_relation_type(project, relation: BaseRelation) -> Optional[str]:
 ) }}
 select * from stream {{ ref('my_seed') }}
 """
+
+streaming_table_schema = """
+version: 2
+
+models:
+  - name: my_streaming_table
+    columns:
+      - name: id
+        description: "The unique identifier for each record"
+        constraints:
+          - type: not_null
+      - name: value
+    config:
+      persist_docs:
+        relation: true
+        columns: true
+"""
+
+
+# There is a single CSV file in the source directory. The contents exactly matches MY_SEED
+streaming_table_from_file = """
+{{ config(
+    materialized='streaming_table',
+) }}
+select * from stream read_files(
+    '{{ env_var('DBT_DATABRICKS_LOCATION_ROOT') }}/test_inputs/streaming_table_test_sources'
+);
+"""

tests/functional/adapter/streaming_tables/test_st_basic.py

@@ -9,9 +9,7 @@
 from tests.functional.adapter.streaming_tables import fixtures
 
 
-@pytest.mark.dlt
-@pytest.mark.skip_profile("databricks_cluster", "databricks_uc_cluster")
-class TestStreamingTablesBasic:
+class TestStreamingTablesMixin:
     @staticmethod
     def insert_record(project, table: BaseRelation, record: tuple[int, int]):
         project.run_sql(f"insert into {table} values {record}")
@@ -38,6 +36,7 @@ def models(self):
             "my_table.sql": MY_TABLE,
             "my_view.sql": MY_VIEW,
             "my_streaming_table.sql": fixtures.streaming_table,
+            "schema.yml": fixtures.streaming_table_schema,
         }
 
     @pytest.fixture(scope="class")
@@ -122,9 +121,29 @@ def setup(self, project, my_streaming_table):
         util.set_model_file(project, my_streaming_table, initial_model)
         project.run_sql(f"drop schema if exists {project.test_schema} cascade")
 
+
+@pytest.mark.dlt
+@pytest.mark.skip_profile("databricks_cluster", "databricks_uc_cluster")
+class TestStreamingTablesBasic(TestStreamingTablesMixin):
     def test_streaming_table_create(self, project, my_streaming_table):
         # setup creates it; verify it's there
         assert self.query_relation_type(project, my_streaming_table) == "streaming_table"
+        # verify the non-null constraint and column comment are persisted on create
+        results = project.run_sql(
+            f"""
+            SELECT
+                is_nullable,
+                comment
+            FROM {project.database}.information_schema.columns
+            WHERE table_catalog = '{project.database}'
+              AND table_schema = '{project.test_schema}'
+              AND table_name = '{my_streaming_table.identifier}'
+              AND column_name = 'id'""",
+            fetch="all",
+        )
+        row = results[0]
+        assert row[0] == "NO"
+        assert row[1] == "The unique identifier for each record"
 
     def test_streaming_table_create_idempotent(self, project, my_streaming_table):
         # setup creates it once; verify it's there and run once
@@ -208,3 +227,34 @@ def test_streaming_table_only_updates_after_refresh(self, project, my_streaming_
         # view until it was refreshed
         assert table_start < table_mid == table_end
         assert view_start == view_mid < view_end
+
+
+@pytest.mark.dlt
+@pytest.mark.skip_profile("databricks_cluster", "databricks_uc_cluster")
+class TestStreamingTablesFromFiles(TestStreamingTablesMixin):
+    @pytest.fixture(scope="class", autouse=True)
+    def models(self):
+        yield {
+            "my_streaming_table.sql": fixtures.streaming_table_from_file,
+            "schema.yml": fixtures.streaming_table_schema,
+        }
+
+    def test_streaming_table_create_from_file(self, project, my_streaming_table):
+        # setup creates it; verify it's there
+        assert self.query_relation_type(project, my_streaming_table) == "streaming_table"
+        # verify the non-null constraint and column comment are persisted on create
+        results = project.run_sql(
+            f"""
+            SELECT
+                is_nullable,
+                comment
+            FROM {project.database}.information_schema.columns
+            WHERE table_catalog = '{project.database}'
+                AND table_schema = '{project.test_schema}'
+                AND table_name = '{my_streaming_table.identifier}'
+                AND column_name = 'id'""",
+            fetch="all",
+        )
+        row = results[0]
+        assert row[0] == "NO"
+        assert row[1] == "The unique identifier for each record"

tests/functional/adapter/streaming_tables/test_st_changes.py

@@ -134,6 +134,10 @@ def project_config_update(self):
     def test_change_is_applied_via_alter(self, project, my_streaming_table):
         self.check_start_state(project, my_streaming_table)
 
+        # This tries to update column definitions (e.g. comment) but should be ignored for
+        # streaming tables. No explicit assertion needed, the job succeeding is sufficient
+        util.write_file(fixtures.streaming_table_schema, "models", "schema.yml")
+
         self.change_config_via_alter(project, my_streaming_table)
         _, logs = util.run_dbt_and_capture(["--debug", "run", "--models", my_streaming_table.name])