Merge 1.13.latest: persist_docs warnings, MV schema drift, dead event cleanup

CHANGELOG.md

@@ -1,3 +1,15 @@
+## dbt-databricks next
+
+### Fixes
+
+- Recreate materialized views when query schema drifts, honoring `on_configuration_change` ([#1621](https://github.com/databricks/dbt-databricks/pull/1621) resolves [#1359](https://github.com/databricks/dbt-databricks/issues/1359))
+- Warn when documented columns are missing from V1 and V2 table and incremental models, and honor `persist_docs.columns` for V2 column comments ([#1563](https://github.com/databricks/dbt-databricks/pull/1563) ports [dbt-adapters#1684](https://github.com/dbt-labs/dbt-adapters/pull/1684), resolving [dbt-adapters#1690](https://github.com/dbt-labs/dbt-adapters/issues/1690)).
+- Extend the missing-column `persist_docs` warning to the view, materialized-view, and streaming-table create paths (post-build validation against the actual relation, gated on `persist_docs.columns`), completing the create-time coverage left as a follow-up to [#1563](https://github.com/databricks/dbt-databricks/pull/1563) ([#1615](https://github.com/databricks/dbt-databricks/pull/1615)).
+
+### Under the Hood
+
+- Remove unused internal logging-event classes (`CredentialLoadError`/`CredentialSaveError`/`CredentialShardEvent`, `PipelineEvent`/`PipelineRefresh`/`PipelineRefreshError`, and the `ConnectionReset`/`ConnectionReuse`/`ConnectionIdleClose`/`ConnectionCreated` connection events) that have had no call sites since the cursor-management and pipeline refactors ([#1547](https://github.com/databricks/dbt-databricks/pull/1547))
+
 ## dbt-databricks 1.12.4 (Aug 12, 2026)
 
 ### Fixes

dbt/adapters/databricks/events/connection_events.py

@@ -33,26 +33,6 @@ def __str__(self) -> str:
         return f"{self.description} - {self.message}"
 
 
-class ConnectionReset(ConnectionWrapperEvent):
-    def __init__(self, description: str):
-        super().__init__(description, "Reset connection handle")
-
-
-class ConnectionReuse(ConnectionWrapperEvent):
-    def __init__(self, description: str, prior_name: str):
-        super().__init__(description, f"Reusing connection previously named {prior_name}")
-
-
 class ConnectionCreate(ConnectionWrapperEvent):
     def __init__(self, description: str):
         super().__init__(description, "Creating connection")
-
-
-class ConnectionIdleClose(ConnectionWrapperEvent):
-    def __init__(self, description: str):
-        super().__init__(description, "Recreating due to idleness")
-
-
-class ConnectionCreated(ConnectionWrapperEvent):
-    def __init__(self, description: str):
-        super().__init__(description, "Connection created")

dbt/adapters/databricks/relation_configs/column_comments.py

@@ -24,7 +24,9 @@ def get_diff(self, other: "ColumnCommentsConfig") -> Optional["ColumnCommentsCon
             other_comments_lower = {k.lower(): v for k, v in other.comments.items()}
 
             for column_name, comment in self.comments.items():
-                # Use case-insensitive comparison for column names
+                # Missing columns are reported post-build.
+                if column_name.lower() not in other_comments_lower:
+                    continue
                 other_comment = other_comments_lower.get(column_name.lower())
                 if comment != other_comment:
                     column_name = f"`{column_name}`"
@@ -53,7 +55,7 @@ def from_relation_config(cls, relation_config: RelationConfig) -> ColumnComments
         columns = getattr(relation_config, "columns", {})
         persist = False
         if relation_config.config:
-            persist = relation_config.config.persist_docs.get("relation") or False
+            persist = relation_config.config.persist_docs.get("columns") or False
         comments = {}
         for column_name, column in columns.items():
             if hasattr(column, "description"):

dbt/include/databricks/macros/adapters/persist_docs.sql

@@ -37,11 +37,43 @@
   {% endif %}
   {% if for_columns and config.persist_column_docs() and model.columns %}
     {%- set existing_columns = adapter.get_columns_in_relation(relation) -%}
-    {%- set columns_to_persist_docs = adapter.get_persist_doc_columns(existing_columns, model.columns) -%}
+    {%- set existing_column_names = existing_columns | map(attribute='name') | list -%}
+    {%- set valid_columns = dbt_databricks_validate_doc_columns(relation, model.columns, existing_column_names) -%}
+    {%- set columns_to_persist_docs = adapter.get_persist_doc_columns(existing_columns, valid_columns) -%}
     {{ alter_column_comment(relation, columns_to_persist_docs) }}
   {% endif %}
 {% endmacro %}
 
+{#-- Match column names case-insensitively. --#}
+{% macro dbt_databricks_validate_doc_columns(relation, column_dict, existing_column_names) -%}
+  {%- set existing_lower = existing_column_names | map('lower') | list -%}
+  {%- set missing = [] -%}
+  {%- set valid = {} -%}
+  {%- for column_name in column_dict -%}
+    {%- if (column_name | lower) in existing_lower -%}
+      {%- do valid.update({column_name: column_dict[column_name]}) -%}
+    {%- else -%}
+      {%- do missing.append(column_name) -%}
+    {%- endif -%}
+  {%- endfor -%}
+  {%- if missing -%}
+    {%- do exceptions.warn(
+      "In relation " ~ relation.render() ~ ": The following columns are specified in the schema "
+      ~ "but are not present in the database: " ~ missing | join(", ")
+    ) -%}
+  {%- endif -%}
+  {{- return(valid) -}}
+{%- endmacro %}
+
+{#-- Validate V2 column docs post-build. --#}
+{% macro validate_persist_doc_columns(relation, model) -%}
+  {% if config.persist_column_docs() and model.columns %}
+    {%- set existing_columns = adapter.get_columns_in_relation(relation) -%}
+    {%- set existing_column_names = existing_columns | map(attribute='name') | list -%}
+    {%- do dbt_databricks_validate_doc_columns(relation, model.columns, existing_column_names) -%}
+  {% endif %}
+{%- endmacro %}
+
 {% macro alter_relation_comment_sql(relation, description) %}
 COMMENT ON {{ relation.type.render().upper() }} {{ relation.render() }} IS '{{ description | replace("'", "\\'") }}'
 {% endmacro %}
@@ -53,4 +85,4 @@ COMMENT ON {{ relation.type.render().upper() }} {{ relation.render() }} IS '{{ d
     {% set column_path = relation.render() ~ '.' ~ adapter.quote(column) %}
     {{ run_query_as(comment_on_column_sql(column_path, escaped_comment), 'main', fetch_result=False) }}
   {% endfor %}
-{% endmacro %}
+{% endmacro %}

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

@@ -81,6 +81,9 @@
       {%- endif -%}
     {%- endif -%}
 
+    {#-- Validate every create, replace, and merge path. --#}
+    {% do validate_persist_doc_columns(target_relation, model) %}
+
     {% set should_revoke = should_revoke(existing_relation, full_refresh_mode) %}
     {% do apply_grants(target_relation, grant_config, should_revoke) %}
     {% do optimize(target_relation) %}

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

@@ -33,9 +33,16 @@
         -- get config options
         {% set on_configuration_change = config.get('on_configuration_change') %}
         {% set configuration_changes = get_configuration_changes(existing_relation) %}
+        {#- Schema drift is a configuration change the components cannot see, so it joins them
+            here and is then subject to the same `on_configuration_change` handling. REFRESH
+            cannot reconcile a drifted schema, so it demands a full refresh. -#}
+        {% set schema_drifted = dlt_inferred_query_schema_changed(existing_relation, sql) %}
+        {%- if schema_drifted and configuration_changes is not none -%}
+            {%- set configuration_changes = configuration_changes.model_copy(update={'requires_full_refresh': true}) -%}
+        {%- endif -%}
 
         {# Skip manual REFRESH on no-op re-runs for auto-refreshed modes. #}
-        {% if configuration_changes is none %}
+        {% if configuration_changes is none and not schema_drifted %}
             {%- set refresh = adapter.get_config_from_model(config.model).config["refresh"] -%}
             {%- if refresh.auto_refreshed -%}
                 {% set build_sql = '' %}
@@ -44,7 +51,12 @@
             {%- endif -%}
 
         {% elif on_configuration_change == 'apply' %}
-            {% set build_sql = get_alter_materialized_view_as_sql(target_relation, configuration_changes, sql, existing_relation, None, None) %}
+            {%- if configuration_changes is none -%}
+                {#- Drift alone: no component changes to alter, so replace outright. -#}
+                {% set build_sql = get_replace_sql(existing_relation, target_relation, sql) %}
+            {%- else -%}
+                {% set build_sql = get_alter_materialized_view_as_sql(target_relation, configuration_changes, sql, existing_relation, None, None) %}
+            {%- endif -%}
         {% elif on_configuration_change == 'continue' %}
             {% set build_sql = "" %}
             {{ exceptions.warn("Configuration changes were identified and `on_configuration_change` was set to `continue` for `" ~ target_relation ~ "`") }}
@@ -74,6 +86,8 @@
 
     {{ execute_multiple_statements(build_sql) }}
 
+    {% do validate_persist_doc_columns(target_relation, model) %}
+
     {%- do apply_tags(target_relation, tags) -%}
 
     {% set column_tags = adapter.get_column_tags_from_model(config.model) %}

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

@@ -74,6 +74,8 @@
 
     {{ execute_multiple_statements(build_sql) }}
 
+    {% do validate_persist_doc_columns(target_relation, model) %}
+
     {%- do apply_tags(target_relation, tags) -%}
 
     {% set should_revoke = should_revoke(existing_relation, full_refresh_mode=True) %}

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

@@ -32,6 +32,8 @@
       {% endif %}
     {% endif %}
 
+    {% do validate_persist_doc_columns(target_relation, model) %}
+
     {% set should_revoke = should_revoke(existing_relation, full_refresh_mode=True) %}
     {{ apply_grants(target_relation, grant_config, should_revoke) }}
     {% do optimize(target_relation) %}

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

@@ -36,6 +36,7 @@
         {{ apply_column_tags(target_relation, column_tags) }}
       {% endif %}
     {% endif %}
+    {% do validate_persist_doc_columns(target_relation, model) %}
     {% set should_revoke = should_revoke(exists_as_view, full_refresh_mode=True) %}
     {% do apply_grants(target_relation, grant_config, should_revoke=True) %}
 
@@ -66,6 +67,8 @@
       {{ apply_column_tags(target_relation, column_tags) }}
     {% endif %}
 
+    {% do validate_persist_doc_columns(target_relation, model) %}
+
     {{ run_hooks(post_hooks) }}
   {% endif %}
 

dbt/include/databricks/macros/relations/schema_drift.sql

@@ -0,0 +1,19 @@
+{#-- CREATE embeds an explicit column list (Databricks user-specified schema).
+    When that drifts from the current query (e.g. upstream `select *` column add),
+    REFRESH fails — the caller replaces instead (#1359). The model's query text is
+    unchanged in that scenario, so the QueryProcessor component of
+    get_configuration_changes cannot detect it; callers treat the result as an
+    additional configuration change. Names only: type-label variance across DESCRIBE
+    paths would cause spurious recreates. Streaming tables (#1303) are out of scope
+    here: locking columns on CREATE is a separate design question. --#}
+{% macro dlt_inferred_query_schema_changed(relation, sql) -%}
+  {%- set inferred_names = get_columns_in_query(sql) | map('lower') | list -%}
+  {%- set existing_names = adapter.get_columns_in_relation(relation)
+                             | map(attribute='name') | map('lower') | list -%}
+  {%- set drifted = inferred_names != existing_names -%}
+  {%- if drifted -%}
+    {%- do log("Materialized view " ~ relation ~ " query schema drifted (was " ~ existing_names
+               ~ ", now " ~ inferred_names ~ "); recreating instead of refreshing.") -%}
+  {%- endif -%}
+  {%- do return(drifted) -%}
+{%- endmacro %}

tests/functional/adapter/materialized_view_tests/fixtures.py

@@ -272,3 +272,48 @@ def materialized_view_with_every(every_value: str) -> str:
 ) }}
 select * from {{ ref('mv_norebuild_seed') }}
 """
+
+# Issue #1359: upstream select * schema evolution against an existing MV.
+schema_evolution_base_v1_sql = """
+{{ config(materialized='table') }}
+select 1 as id, 'foo' as name
+"""
+
+schema_evolution_base_v2_sql = """
+{{ config(materialized='table') }}
+select 1 as id, 'foo' as name, 42 as new_column
+"""
+
+schema_evolution_mv_sql = """
+{{ config(materialized='materialized_view', on_configuration_change='apply') }}
+select * from {{ ref('schema_evolution_base') }}
+"""
+
+schema_evolution_mv_yml = """
+version: 2
+models:
+  - name: schema_evolution_mv
+    columns:
+      - name: id
+      - name: name
+"""
+
+schema_evolution_mv_yml_v2 = """
+version: 2
+models:
+  - name: schema_evolution_mv
+    columns:
+      - name: id
+      - name: name
+      - name: new_column
+"""
+
+schema_evolution_mv_fail_sql = """
+{{ config(materialized='materialized_view', on_configuration_change='fail') }}
+select * from {{ ref('schema_evolution_base') }}
+"""
+
+schema_evolution_mv_continue_sql = """
+{{ config(materialized='materialized_view', on_configuration_change='continue') }}
+select * from {{ ref('schema_evolution_base') }}
+"""