diff --git a/CHANGELOG.md b/CHANGELOG.md index 2e8f9289e..f97ca715c 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/dbt/include/databricks/macros/materializations/materialized_view.sql b/dbt/include/databricks/macros/materializations/materialized_view.sql index 0ed13fd03..a7859052b 100644 --- a/dbt/include/databricks/macros/materializations/materialized_view.sql +++ b/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 ~ "`") }} diff --git a/dbt/include/databricks/macros/relations/schema_drift.sql b/dbt/include/databricks/macros/relations/schema_drift.sql new file mode 100644 index 000000000..18df7478e --- /dev/null +++ b/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 %} diff --git a/tests/functional/adapter/materialized_view_tests/fixtures.py b/tests/functional/adapter/materialized_view_tests/fixtures.py index 00450ad06..bf34f5dc2 100644 --- a/tests/functional/adapter/materialized_view_tests/fixtures.py +++ b/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') }} +""" diff --git a/tests/functional/adapter/materialized_view_tests/test_mv_schema_evolution.py b/tests/functional/adapter/materialized_view_tests/test_mv_schema_evolution.py new file mode 100644 index 000000000..bf23299be --- /dev/null +++ b/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")] diff --git a/tests/unit/macros/relations/test_schema_drift.py b/tests/unit/macros/relations/test_schema_drift.py new file mode 100644 index 000000000..87c09c868 --- /dev/null +++ b/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