fix: use CREATE OR REPLACE for view-to-view replaces in safer mode

CHANGELOG.md

@@ -1,3 +1,9 @@
+## dbt-databricks 1.12.8 (TBD)
+
+### Fixes
+
+- Replace a view with `CREATE OR REPLACE VIEW` when `use_safer_relation_operations` is on, so a view whose upstream dropped or renamed a column it used no longer fails with `UNRESOLVED_COLUMN` (thanks @shantanutheone!) ([#1733](https://github.com/databricks/dbt-databricks/pull/1733) resolves [#1714](https://github.com/databricks/dbt-databricks/issues/1714))
+
 ## dbt-databricks 1.12.7 (Oct 12, 2026)
 
 ### Fixes

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

@@ -34,11 +34,14 @@
   {% set file_format = adapter.resolve_file_format(config) %}
   {% set is_replaceable = existing_relation.type == target_relation.type and existing_relation.can_be_replaced and file_format == "delta" %}
 
+  {#- Renaming a view makes Databricks re-resolve its old query, which fails once an upstream column is gone, and CREATE OR REPLACE VIEW is already atomic. -#}
+  {% if is_replaceable and existing_relation.is_view %}
+    {{ return(get_replace_view_sql(target_relation, sql)) }}
+  {% endif %}
+
   {% if not safe_replace %}
     {# Prioritize 'create or replace' for speed #}
-    {% if is_replaceable and existing_relation.is_view %}
-      {{ return(get_replace_view_sql(target_relation, sql)) }}
-    {% elif is_replaceable and existing_relation.is_table %}
+    {% if is_replaceable and existing_relation.is_table %}
       {{ return(get_replace_table_sql(target_relation, sql)) }}
     {% elif is_replaceable and existing_relation.is_materialized_view %}
       {{ return(get_replace_materialized_view_sql(target_relation, sql)) }}

docs/flow/replace_flow.md

@@ -1,6 +1,6 @@
 # Replace Flow
 
-_Last updated: 2026-10-09_
+_Last updated: 2026-10-18_
 
 Shared decision tree used when view, materialized-view, streaming-table, or metric-view helpers must
 replace an existing relation. Table and incremental V2 use their dedicated
@@ -19,11 +19,12 @@ flowchart TD
     TABLE -- yes --> ERROR[Raise not-implemented compiler error]
     TABLE -- no --> METRIC{Target and existing are both metric views?}
     METRIC -- yes --> METRICSQL[get_replace_metric_view_sql]
-    METRIC -- no --> SAFE{use_safer_relation_operations?}
+    METRIC -- no --> VIEW{"Existing and target are views, existing can be replaced,\nand configured file format is Delta?"}
+    VIEW -- yes --> VIEWSQL[get_replace_view_sql]
+    VIEW -- no --> SAFE{use_safer_relation_operations?}
 
     SAFE -- false --> DIRECT{"Same relation type, existing can be replaced,\nand configured file format is Delta?"}
     DIRECT -- yes --> TYPE{Existing relation type?}
-    TYPE -- view --> VIEWSQL[get_replace_view_sql]
     TYPE -- "materialized view" --> MVSQL[get_replace_materialized_view_sql]
     DIRECT -- no --> TARGETRENAME{Target can be renamed?}
     TYPE -- other --> TARGETRENAME
@@ -37,10 +38,14 @@ flowchart TD
     EXISTRENAME2 -- no --> DROP["drop_and_create:<br/>drop existing; create target"]
 ```
 
-Direct replacement is available only when safe operations are disabled and the existing and target
-relations have the same type, the existing relation is replaceable, and the configured file format
-is Delta. In practice, the supported direct branches here are views and materialized views; table
-targets have already failed at the initial guard.
+A view replacing a replaceable view always uses `CREATE OR REPLACE VIEW`, with or without
+`use_safer_relation_operations`. The statement is atomic and a view holds no data, so a backup adds
+nothing, and renaming the old view fails once an upstream relation has dropped a column it selects.
+
+Other direct replacement is available only when safe operations are disabled and the existing and
+target relations have the same type, the existing relation is replaceable, and the configured file
+format is Delta. In practice, the remaining direct branch here is materialized views; table targets
+have already failed at the initial guard.
 
 Tables and views can be renamed, except that an existing Hive metastore table is never renamed:
 Databricks rejects `ALTER TABLE ... RENAME TO` for managed Delta tables on S3. A view replacing a

tests/functional/adapter/relations/fixtures.py

@@ -23,6 +23,26 @@
 select 1 as id
 """
 
+safer_ops_upstream_sql = """
+{{ config(materialized='table') }}
+select 1 as id, 'a' as old_col
+"""
+
+safer_ops_upstream_renamed_sql = """
+{{ config(materialized='table') }}
+select 1 as id, 'a' as new_col
+"""
+
+safer_ops_view_sql = """
+{{ config(materialized='view') }}
+select id, old_col from {{ ref('safer_ops_upstream') }}
+"""
+
+safer_ops_view_renamed_sql = """
+{{ config(materialized='view') }}
+select id, new_col from {{ ref('safer_ops_upstream') }}
+"""
+
 replaced_relation_source_sql = """
 {{ config(materialized='table') }}
 select 1 as id union all select 2 as id union all select 3 as id

tests/functional/adapter/relations/test_safe_relation_operations.py

@@ -4,6 +4,8 @@
   breaks history/lineage.
 * incremental.sql reads the config with `| as_bool`, so a YAML string "false" is correctly
   treated as False (in-place, stable Delta id).
+* A view is replaced even after its upstream dropped a column the old view selects, because
+  view-to-view replaces use CREATE OR REPLACE VIEW instead of renaming the old view.
 """
 
 import pytest
@@ -72,3 +74,34 @@ def test_string_false_does_not_swap_on_incremental(self, project):
             "incremental reads use_safer_relation_operations | as_bool; string 'false' should "
             f"be False and not swap; Delta id changed {first_id} -> {second_id}"
         )
+
+
+@pytest.mark.skip_profile("databricks_cluster")
+class TestSaferOpsReplacesViewAfterUpstreamColumnRename(RerunSafeMixin, MaterializationV2Mixin):
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "safer_ops_upstream.sql": fixtures.safer_ops_upstream_sql,
+            "safer_ops_view.sql": fixtures.safer_ops_view_sql,
+        }
+
+    @pytest.fixture(scope="class")
+    def relations_to_reset(self):
+        return ("safer_ops_view", "safer_ops_upstream")
+
+    @pytest.fixture(scope="class")
+    def project_config_update(self):
+        return {
+            "flags": {"use_materialization_v2": True},
+            "models": {"+use_safer_relation_operations": True},
+        }
+
+    def test_view_is_replaced(self, project):
+        util.run_dbt(["run"])
+        util.write_file(fixtures.safer_ops_upstream_renamed_sql, "models", "safer_ops_upstream.sql")
+        util.write_file(fixtures.safer_ops_view_renamed_sql, "models", "safer_ops_view.sql")
+        util.run_dbt(["run"])
+        rows = project.run_sql(
+            "select id, new_col from {database}.{schema}.safer_ops_view", fetch="all"
+        )
+        assert [tuple(row) for row in rows] == [(1, "a")]

tests/unit/macros/relations/test_replace_macros.py

@@ -82,19 +82,6 @@ def test_unity_catalog_table_to_view_backs_up_table(
         assert statements == ["rename model__dbt_stg", "drop backup"]
         context["create_backup"].assert_called_once_with(existing)
 
-    def test_hive_metastore_view_to_view_with_safe_replace_backs_up_view(
-        self, template_bundle, config, context
-    ):
-        config["use_safer_relation_operations"] = True
-        existing = DatabricksRelation.create(
-            database="hive_metastore", schema="schema", identifier="model", type="view"
-        )
-
-        statements = self.get_replace_sql(template_bundle, existing, "view")
-
-        assert statements == ["rename model__dbt_stg", "drop backup"]
-        context["create_backup"].assert_called_once_with(existing)
-
     @pytest.mark.parametrize("target_type", ["materialized_view", "streaming_table"])
     def test_unity_catalog_table_to_non_renamable_backs_up_table(
         self, template_bundle, context, target_type
@@ -108,16 +95,19 @@ def test_unity_catalog_table_to_non_renamable_backs_up_table(
         assert statements == [f"create {target_type}", "drop backup"]
         context["create_backup"].assert_called_once_with(existing)
 
+    @pytest.mark.parametrize("safe_replace", [False, True])
     @pytest.mark.parametrize("database", ["hive_metastore", "main"])
-    def test_view_to_view_without_safe_replace_uses_create_or_replace(
-        self, template_bundle, context, database
+    def test_view_to_view_uses_create_or_replace(
+        self, template_bundle, config, context, database, safe_replace
     ):
+        config["use_safer_relation_operations"] = safe_replace
         existing = DatabricksRelation.create(
             database=database, schema="schema", identifier="model", type="view"
         )
 
         assert self.get_replace_sql(template_bundle, existing, "view") == "create or replace view"
         context["create_backup"].assert_not_called()
+        context["drop_relation_if_exists"].assert_not_called()
 
 
 def model_relation(type):