Fix: Handle view alterations and existing table replacements

This commit addresses the following:

- Modifies the `create view` macro to use `create or alter view` when the view already exists, preventing errors during re-materialization.
- Updates the view materialization logic to correctly avoid renaming if no table exists.
- Adds test cases to verify that materializing an existing view updates its definition correctly, including column changes, previous grants and avoids table changes if view fails.

Author: axellpadilla

dbt/include/sqlserver/macros/materializations/models/view/view.sql

@@ -26,6 +26,13 @@
   {%- set preexisting_backup_relation = load_cached_relation(backup_relation) -%}
   -- grab current tables grants config for comparision later on
   {% set grant_config = config.get('grants') %}
+  {% set preserved_grants = {} %}
+
+  {% if existing_relation is not none and existing_relation.type != 'view' %}
+    {% set current_grants_table = run_query(get_show_grant_sql(existing_relation)) %}
+    {% set current_grants_dict = adapter.standardize_grants_dict(current_grants_table) %}
+    {% set preserved_grants = diff_of_two_dicts(current_grants_dict, grant_config) %}
+  {% endif %}
 
   {{ run_hooks(pre_hooks, inside_transaction=False) }}
 
@@ -36,26 +43,34 @@
   -- `BEGIN` happens here:
   {{ run_hooks(pre_hooks, inside_transaction=True) }}
 
-  -- build model
-  {% call statement('main') -%}
-    {{ get_create_view_as_sql(intermediate_relation, sql) }}
-  {%- endcall %}
+  {% if existing_relation is not none and existing_relation.type != 'view' %}
+    -- build model
+    {% call statement('main') -%}
+      {{ get_create_view_as_sql(intermediate_relation, sql) }}
+    {%- endcall %}
 
-  -- cleanup
-  -- move the existing view out of the way
-  {% if existing_relation is not none %}
-     /* Do the equivalent of rename_if_exists. 'existing_relation' could have been dropped
-        since the variable was first set. */
+    -- cleanup
+    -- move the existing relation out of the way
     {% set existing_relation = load_cached_relation(existing_relation) %}
     {% if existing_relation is not none %}
         {{ adapter.rename_relation(existing_relation, backup_relation) }}
     {% endif %}
+
+    {{ adapter.rename_relation(intermediate_relation, target_relation) }}
+  {% else %}
+    -- build model
+    {% call statement('main') -%}
+      {{ get_create_view_as_sql(target_relation, sql) }}
+    {%- endcall %}
   {% endif %}
-  {{ adapter.rename_relation(intermediate_relation, target_relation) }}
 
   {% set should_revoke = should_revoke(existing_relation, full_refresh_mode=True) %}
   {% do apply_grants(target_relation, grant_config, should_revoke=should_revoke) %}
 
+  {% if preserved_grants %}
+    {% do apply_grants(target_relation, preserved_grants, should_revoke=False) %}
+  {% endif %}
+
   {% do persist_docs(target_relation, model) %}
 
   {{ run_hooks(post_hooks, inside_transaction=True) }}

dbt/include/sqlserver/macros/relations/views/create.sql

@@ -7,7 +7,7 @@
     {%- endif %}
 
     {% set query %}
-        create view {{ relation.include(database=False) }} as {{ sql }};
+        CREATE OR ALTER VIEW {{ relation.include(database=False) }} AS {{ sql }};
     {% endset %}
 
     {% set tst %}

tests/functional/adapter/dbt/test_basic.py

@@ -1,3 +1,5 @@
+import os
+
 import pytest
 from dbt.tests.adapter.basic.test_adapter_methods import BaseAdapterMethod
 from dbt.tests.adapter.basic.test_base import BaseSimpleMaterializations
@@ -9,10 +11,64 @@
 from dbt.tests.adapter.basic.test_singular_tests_ephemeral import BaseSingularTestsEphemeral
 from dbt.tests.adapter.basic.test_snapshot_check_cols import BaseSnapshotCheckCols
 from dbt.tests.adapter.basic.test_snapshot_timestamp import BaseSnapshotTimestamp
+from dbt.tests.util import run_dbt
 
 
 class TestSimpleMaterializations(BaseSimpleMaterializations):
-    pass
+
+    def test_existing_view_materialization(self, project, models):
+        """Test that materializing an existing view works correctly."""
+        # Create a temporary model file directly in the project
+        model_path = os.path.join(project.project_root, "models", "view_model_exists.sql")
+
+        # Write the initial model without the value column
+        with open(model_path, "w") as f:
+            f.write(
+                """
+                {{ config(materialized='view') }}
+                select
+                1 as id
+                {% if var('include_value_column', false) %}
+                , 2 as value
+                {% endif %}
+            """
+            )
+
+        # First run to create the view without the extra column
+        results = run_dbt(["run", "-m", "view_model_exists"])
+        assert len(results) == 1
+
+        # Generate catalog to get column information
+        catalog = run_dbt(["docs", "generate"])
+
+        # Check columns in the catalog
+        node_id = "model.base.view_model_exists"
+        assert node_id in catalog.nodes
+
+        # Get columns from the catalog
+        columns = catalog.nodes[node_id].columns
+        column_names = [name.lower() for name in columns.keys()]
+
+        # Verify only the id column exists
+        assert "id" in column_names
+        assert "value" not in column_names
+
+        # Second run with a variable to include the extra column
+        results = run_dbt(
+            ["run", "-m", "view_model_exists", "--vars", '{"include_value_column": true}']
+        )
+        assert len(results) == 1
+
+        # Generate catalog again to get updated column information
+        catalog = run_dbt(["docs", "generate"])
+
+        # Get updated columns from the catalog
+        columns = catalog.nodes[node_id].columns
+        column_names = [name.lower() for name in columns.keys()]
+
+        # Verify both columns exist now
+        assert "id" in column_names
+        assert "value" in column_names
 
 
 class TestSingularTests(BaseSingularTests):

tests/functional/adapter/dbt/test_constraints.py

@@ -471,7 +471,7 @@ def models(self):
     @pytest.fixture(scope="class")
     def expected_sql(self):
         return """
-        EXEC(' create view <model_identifier> as -- depends_on: <foreign_key_model_identifier> select ''blue'' as color, 1 as id, ''2019-01-01'' as date_day; ') EXEC(' CREATE TABLE <model_identifier> ( id int not null , color varchar(100), date_day varchar(100) ) INSERT INTO <model_identifier> ( [id], [color], [date_day] ) SELECT [id], [color], [date_day] FROM <model_identifier>  ') EXEC('DROP VIEW IF EXISTS <model_identifier>
+      EXEC(' CREATE OR ALTER VIEW <model_identifier> AS -- depends_on: <foreign_key_model_identifier> select ''blue'' as color, 1 as id, ''2019-01-01'' as date_day; ') EXEC(' CREATE TABLE <model_identifier> ( id int not null , color varchar(100), date_day varchar(100) ) INSERT INTO <model_identifier> ( [id], [color], [date_day] ) SELECT [id], [color], [date_day] FROM <model_identifier>  ') EXEC('DROP VIEW IF EXISTS <model_identifier>
         """
 
     # EXEC('DROP view IF EXISTS <model_identifier>
@@ -591,7 +591,7 @@ def models(self):
     @pytest.fixture(scope="class")
     def expected_sql(self):
         return """
-        EXEC(' create view <model_identifier> as -- depends_on: <foreign_key_model_identifier> select ''blue'' as color, 1 as id, ''2019-01-01'' as date_day; ') EXEC(' CREATE TABLE <model_identifier> ( id int not null , color varchar(100), date_day varchar(100) ) INSERT INTO <model_identifier> ( [id], [color], [date_day] ) SELECT [id], [color], [date_day] FROM <model_identifier>  ') EXEC('DROP VIEW IF EXISTS <model_identifier>
+      EXEC(' CREATE OR ALTER VIEW <model_identifier> AS -- depends_on: <foreign_key_model_identifier> select ''blue'' as color, 1 as id, ''2019-01-01'' as date_day; ') EXEC(' CREATE TABLE <model_identifier> ( id int not null , color varchar(100), date_day varchar(100) ) INSERT INTO <model_identifier> ( [id], [color], [date_day] ) SELECT [id], [color], [date_day] FROM <model_identifier>  ') EXEC('DROP VIEW IF EXISTS <model_identifier>
         """
 
     def test__model_constraints_ddl(self, project, expected_sql):

tests/functional/adapter/mssql/test_materialize_change.py

@@ -23,6 +23,15 @@
 SELECT 1 AS data
 """
 
+invalid_view_mat = """
+{{
+    config({
+    "materialized": 'view'
+    })
+}}
+SELECT * FROM missing_relation
+"""
+
 schema = """
 version: 2
 models:
@@ -50,6 +59,57 @@ def test_passes(self, project):
         run_dbt(["run"])
 
 
+class TestTabletoViewRollback(BaseTableView):
+    """Test that a failed table to view replacement leaves the original table intact."""
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {"mat_object.sql": invalid_view_mat, "schema.yml": schema}
+
+    def test_existing_table_is_preserved(self, project):
+        self.create_object(
+            project, f"SELECT * INTO {project.test_schema}.mat_object FROM ({model_sql}) t"
+        )
+
+        failing_results = run_dbt(["run"], expect_pass=False)
+        assert len(failing_results) == 1
+
+        rows = project.run_sql(f"select * from {project.test_schema}.mat_object", fetch="all")
+        assert len(rows) == 1
+        assert rows[0][0] == 1
+
+
+class TestTabletoViewPreservesGrants(BaseTableView):
+    """Test that grants on the existing table are preserved on the replaced view."""
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {"mat_object.sql": view_mat, "schema.yml": schema}
+
+    def test_public_select_grant_survives_swap(self, project):
+        self.create_object(
+            project, f"SELECT * INTO {project.test_schema}.mat_object FROM ({model_sql}) t"
+        )
+        project.run_sql(f"grant select on object::{project.test_schema}.mat_object to public")
+
+        run_dbt(["run"])
+
+        grants = project.run_sql(
+            f"""
+                        select pr.name as grantee, pe.permission_name
+                        from sys.database_permissions pe
+                        join sys.objects o on pe.major_id = o.object_id
+                        join sys.schemas s on o.schema_id = s.schema_id
+                        join sys.database_principals pr
+                            on pe.grantee_principal_id = pr.principal_id
+                        where s.name = '{project.test_schema}'
+                            and o.name = 'mat_object'
+            """,
+            fetch="all",
+        )
+        assert ("PUBLIC", "SELECT") in [(row[0].upper(), row[1].upper()) for row in grants]
+
+
 class TestViewtoTable(BaseTableView):
     """Test if changing from a view object to a table object correctly replaces"""