fix: recreate materialized views on query schema drift

CHANGELOG.md

@@ -2,6 +2,7 @@
 
 ### 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)).
 
 ### Under the Hood

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 ~ "`") }}

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

@@ -257,3 +257,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') }}
+"""

tests/functional/adapter/materialized_view_tests/test_mv_schema_evolution.py

@@ -0,0 +1,103 @@
+"""Schema evolution for materialized views (issue #1359)."""
+
+import pytest
+from dbt.tests import util
+
+from tests.functional.adapter.fixtures import RerunSafeMixin
+from tests.functional.adapter.materialized_view_tests import fixtures
+
+
+@pytest.mark.dlt
+@pytest.mark.skip_profile("databricks_cluster", "databricks_uc_cluster")
+class TestMaterializedViewSchemaEvolution(RerunSafeMixin):
+    """Upstream ``select *`` column adds must recreate the MV, not REFRESH."""
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "schema_evolution_base.sql": fixtures.schema_evolution_base_v1_sql,
+            "schema_evolution_mv.sql": fixtures.schema_evolution_mv_sql,
+            "schema_evolution_mv.yml": fixtures.schema_evolution_mv_yml,
+        }
+
+    @pytest.fixture(scope="class")
+    def relations_to_reset(self):
+        return ("schema_evolution_mv", "schema_evolution_base")
+
+    def test_upstream_column_add_recreates_without_full_refresh_flag(self, project):
+        util.run_dbt(["run"])
+        util.write_file(
+            fixtures.schema_evolution_base_v2_sql, "models", "schema_evolution_base.sql"
+        )
+
+        util.run_dbt(["run"])
+        rows = project.run_sql("select id, name, new_column from schema_evolution_mv", fetch="all")
+        assert rows == [(1, "foo", 42)]
+
+    def test_column_add_to_properties_yaml_recreates(self, project):
+        """The issue's exact sequence: upstream gains a column, then the YAML follows."""
+        util.run_dbt(["run"])
+        util.write_file(
+            fixtures.schema_evolution_base_v2_sql, "models", "schema_evolution_base.sql"
+        )
+        util.write_file(fixtures.schema_evolution_mv_yml_v2, "models", "schema_evolution_mv.yml")
+
+        util.run_dbt(["run"])
+        rows = project.run_sql("select id, name, new_column from schema_evolution_mv", fetch="all")
+        assert rows == [(1, "foo", 42)]
+
+
+@pytest.mark.dlt
+@pytest.mark.skip_profile("databricks_cluster", "databricks_uc_cluster")
+class TestMaterializedViewSchemaDriftOnConfigurationChangeFail(RerunSafeMixin):
+    """Schema drift is a configuration change, so `fail` must stop the run."""
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "schema_evolution_base.sql": fixtures.schema_evolution_base_v1_sql,
+            "schema_evolution_mv.sql": fixtures.schema_evolution_mv_fail_sql,
+        }
+
+    @pytest.fixture(scope="class")
+    def relations_to_reset(self):
+        return ("schema_evolution_mv", "schema_evolution_base")
+
+    def test_drift_fails_and_leaves_mv_untouched(self, project):
+        util.run_dbt(["run"])
+        util.write_file(
+            fixtures.schema_evolution_base_v2_sql, "models", "schema_evolution_base.sql"
+        )
+
+        util.run_dbt(["run"], expect_pass=False)
+
+        rows = project.run_sql("select * from schema_evolution_mv", fetch="all")
+        assert rows == [(1, "foo")]
+
+
+@pytest.mark.dlt
+@pytest.mark.skip_profile("databricks_cluster", "databricks_uc_cluster")
+class TestMaterializedViewSchemaDriftOnConfigurationChangeContinue(RerunSafeMixin):
+    """Schema drift is a configuration change, so `continue` must skip the rebuild."""
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "schema_evolution_base.sql": fixtures.schema_evolution_base_v1_sql,
+            "schema_evolution_mv.sql": fixtures.schema_evolution_mv_continue_sql,
+        }
+
+    @pytest.fixture(scope="class")
+    def relations_to_reset(self):
+        return ("schema_evolution_mv", "schema_evolution_base")
+
+    def test_drift_continues_and_leaves_mv_untouched(self, project):
+        util.run_dbt(["run"])
+        util.write_file(
+            fixtures.schema_evolution_base_v2_sql, "models", "schema_evolution_base.sql"
+        )
+
+        util.run_dbt(["run"])
+
+        rows = project.run_sql("select * from schema_evolution_mv", fetch="all")
+        assert rows == [(1, "foo")]

tests/unit/macros/relations/test_schema_drift.py

@@ -0,0 +1,58 @@
+"""Macro tests for MV query schema-drift detection (issue #1359)."""
+
+import pytest
+
+from dbt.adapters.databricks.column import DatabricksColumn
+from tests.unit.macros.base import MacroTestBase
+
+
+class TestDltInferredQuerySchemaChanged(MacroTestBase):
+    @pytest.fixture(scope="class")
+    def template_name(self) -> str:
+        return "schema_drift.sql"
+
+    @pytest.fixture(scope="class")
+    def macro_folders_to_load(self) -> list:
+        return ["macros", "macros/relations"]
+
+    def detect(self, template_bundle, existing: list[str], inferred: list[str]) -> bool:
+        """Run the macro with the two column-name lists the adapter would return.
+
+        The macro yields its verdict via `do return(...)`, which renders as empty text,
+        so capture the value the harness's `return` hook receives.
+        """
+        template_bundle.context["adapter"].get_columns_in_relation = lambda relation: [
+            DatabricksColumn.create(name, "string") for name in existing
+        ]
+        template_bundle.context["get_columns_in_query"] = lambda sql: inferred
+        returned: list[bool] = []
+        template_bundle.context["return"] = returned.append
+        self.run_macro_raw(
+            template_bundle.template,
+            "dlt_inferred_query_schema_changed",
+            template_bundle.relation,
+            "select * from upstream",
+        )
+        assert len(returned) == 1, f"macro returned {len(returned)} values, expected 1"
+        return returned[0]
+
+    def test_same_names_same_order(self, template_bundle):
+        assert self.detect(template_bundle, ["id", "name"], ["id", "name"]) is False
+
+    def test_case_insensitive_name_match(self, template_bundle):
+        assert self.detect(template_bundle, ["ID", "Name"], ["id", "name"]) is False
+
+    def test_added_column(self, template_bundle):
+        assert self.detect(template_bundle, ["id", "name"], ["id", "name", "new_column"]) is True
+
+    def test_removed_column(self, template_bundle):
+        assert self.detect(template_bundle, ["id", "name", "gone"], ["id", "name"]) is True
+
+    def test_reordered_columns(self, template_bundle):
+        assert self.detect(template_bundle, ["id", "name"], ["name", "id"]) is True
+
+    def test_empty_inferred_columns(self, template_bundle):
+        assert self.detect(template_bundle, ["id"], []) is True
+
+    def test_both_empty(self, template_bundle):
+        assert self.detect(template_bundle, [], []) is False