Fix: Handle view materialization alterations

dbt/include/sqlserver/macros/adapters/metadata.sql

@@ -90,6 +90,20 @@
   {{ return(load_result('get_relation_without_caching').table) }}
 {% endmacro %}
 
+{% macro get_view_definition_sql(relation) %}
+    {{ return(adapter.dispatch('get_view_definition_sql')(relation)) }}
+{% endmacro %}
+
+{% macro sqlserver__get_view_definition_sql(relation) -%}
+  {{ get_use_database_sql(relation.database) }}
+  select object_definition(v.object_id) as definition
+  from sys.views as v {{ information_schema_hints() }}
+  inner join sys.schemas as s {{ information_schema_hints() }}
+    on v.schema_id = s.schema_id
+  where upper(s.name) = upper('{{ relation.schema }}')
+    and upper(v.name) = upper('{{ relation.identifier }}')
+{% endmacro %}
+
 {% macro sqlserver__get_relation_last_modified(information_schema, relations) -%}
   {%- call statement('last_modified', fetch_result=True) -%}
         select

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

@@ -26,6 +26,36 @@
   {%- 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 = {} %}
+  {% set should_skip_view_update = false %}
+  {% set build_sql = none %}
+
+  {% 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 = {} %}
+    {% for privilege, grantees in diff_of_two_dicts(current_grants_dict, grant_config).items() %}
+      {% if privilege | lower in ['select', 'insert', 'update', 'delete'] %}
+        {% do preserved_grants.update({privilege: grantees}) %}
+      {% endif %}
+    {% endfor %}
+    {% set build_sql = get_create_view_as_sql(intermediate_relation, sql) %}
+  {% elif existing_relation is not none and existing_relation.type == 'view' %}
+    {% set current_view_definition_table = run_query(get_view_definition_sql(existing_relation)) %}
+    {% if current_view_definition_table is not none and current_view_definition_table.rows | length > 0 %}
+      {% set normalized_relation = target_relation.include(database=False) | lower | replace('\n', '') | replace('\r', '') | replace('\t', '') | replace(' ', '') | replace(';', '') %}
+      {% set normalized_sql = sql | lower | replace('\n', '') | replace('\r', '') | replace('\t', '') | replace(' ', '') | replace(';', '') %}
+      {% set normalized_definition = current_view_definition_table.rows[0][0] | lower | replace('\n', '') | replace('\r', '') | replace('\t', '') | replace(' ', '') | replace(';', '') %}
+      {% set should_skip_view_update = normalized_definition.endswith(normalized_sql) %}
+    {% endif %}
+    {% if should_skip_view_update %}
+      {% set build_sql = 'declare @dbt_sqlserver_noop int;' %}
+    {% else %}
+      {% set build_sql = get_create_view_as_sql(target_relation, sql) %}
+    {% endif %}
+  {% else %}
+    {% set build_sql = get_create_view_as_sql(target_relation, sql) %}
+  {% endif %}
 
   {{ run_hooks(pre_hooks, inside_transaction=False) }}
 
@@ -36,26 +66,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') -%}
+      {{ build_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') -%}
+      {{ build_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
@@ -10,10 +12,62 @@
 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

@@ -472,7 +472,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> WITH (TABLOCK) ( [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>
@@ -592,7 +592,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> WITH (TABLOCK) ( [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

@@ -24,6 +24,15 @@
 SELECT 1 AS data
 """
 
+invalid_view_mat = """
+{{
+    config({
+    "materialized": 'view'
+    })
+}}
+SELECT * FROM missing_relation
+"""
+
 schema = """
 version: 2
 models:
@@ -51,6 +60,97 @@ 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, insert, update, delete
+                on object::{project.test_schema}.mat_object to public""")
+
+        run_dbt(["run"])
+
+        grant_count = project.run_sql(
+            f"""
+                        select count(*)
+                        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'
+                            and pe.permission_name in ('SELECT', 'INSERT', 'UPDATE', 'DELETE')
+            """,
+            fetch="one",
+        )
+        assert grant_count[0] == 4
+
+
+class TestViewMaterializationNoOp(BaseTableView):
+    """Test that rerunning an unchanged view avoids altering the view."""
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {"mat_object.sql": view_mat, "schema.yml": schema}
+
+    def test_unchanged_view_does_not_alter(self, project):
+        self.create_object(project, f"CREATE VIEW {project.test_schema}.mat_object AS {model_sql}")
+
+        before_modify_date = project.run_sql(
+            f"""
+            select modify_date
+            from sys.objects o
+            join sys.schemas s on o.schema_id = s.schema_id
+            where upper(s.name) = upper('{project.test_schema}')
+              and upper(o.name) = upper('mat_object')
+            """,
+            fetch="one",
+        )[0]
+
+        results = run_dbt(["run"])
+        assert len(results) == 1
+
+        after_modify_date = project.run_sql(
+            f"""
+            select modify_date
+            from sys.objects o
+            join sys.schemas s on o.schema_id = s.schema_id
+            where upper(s.name) = upper('{project.test_schema}')
+              and upper(o.name) = upper('mat_object')
+            """,
+            fetch="one",
+        )[0]
+
+        assert after_modify_date == before_modify_date
+
+
 class TestViewtoTable(BaseTableView):
     """Test if changing from a view object to a table object correctly replaces"""