Update drop all schemas to target only schemas of a table in a given index

dbt/include/sqlserver/macros/adapter/indexes.sql

@@ -108,6 +108,7 @@
         inner join sys.tables {{ information_schema_hints() }}
         on sys.indexes.object_id = sys.tables.object_id
         where sys.indexes.[name] is not null
+        and SCHEMA_NAME(sys.tables.schema_id) = '{{ this.schema }}'
         and sys.tables.[name] = '{{ this.table }}'
         for xml path('')
     ); exec sp_executesql @drop_remaining_indexes_last;

dbt/include/sqlserver/macros/materializations/tests.sql

@@ -0,0 +1,46 @@
+{% macro sqlserver__get_test_sql(main_sql, fail_calc, warn_if, error_if, limit) -%}
+
+  -- Create target schema if it does not
+  USE [{{ target.database }}];
+  IF NOT EXISTS (SELECT * FROM sys.schemas WHERE name = '{{ target.schema }}')
+  BEGIN
+    EXEC('CREATE SCHEMA [{{ target.schema }}]')
+  END
+
+  {% set with_statement_pattern = 'with .+ as\s*\(' %}
+  {% set re = modules.re %}
+  {% set is_match = re.search(with_statement_pattern, main_sql, re.IGNORECASE) %}
+
+  {% if is_match %}
+    {% set testview %}
+      [{{ target.schema }}.testview_{{ range(1300, 19000) | random }}]
+    {% endset %}
+
+    {% set sql = main_sql.replace("'", "''")%}
+    EXEC('create view {{testview}} as {{ sql }};')
+    select
+      {{ "top (" ~ limit ~ ')' if limit != none }}
+      {{ fail_calc }} as failures,
+      case when {{ fail_calc }} {{ warn_if }}
+        then 'true' else 'false' end as should_warn,
+      case when {{ fail_calc }} {{ error_if }}
+        then 'true' else 'false' end as should_error
+    from (
+      select * from {{testview}}
+    ) dbt_internal_test;
+
+    EXEC('drop view {{testview}};')
+
+  {% else -%}
+    select
+      {{ "top (" ~ limit ~ ')' if limit != none }}
+      {{ fail_calc }} as failures,
+      case when {{ fail_calc }} {{ warn_if }}
+        then 'true' else 'false' end as should_warn,
+      case when {{ fail_calc }} {{ error_if }}
+        then 'true' else 'false' end as should_error
+    from (
+      {{ main_sql }}
+    ) dbt_internal_test
+  {%- endif -%}
+{%- endmacro %}

tests/functional/adapter/mssql/test_index.py

@@ -54,6 +54,18 @@
   select * from {{ ref('raw_data') }}
 """
 
+drop_schema_model = """
+{{
+  config({
+  "materialized": 'table',
+        "post-hook": [
+            "{{ drop_all_indexes_on_table() }}",
+        ]
+  })
+}}
+select * from {{ ref('raw_data') }}
+"""
+
 base_validation = """
 with base_query AS (
 select i.[name] as index_name,
@@ -107,6 +119,21 @@
 """
 )
 
+other_index_count = (
+    base_validation
+    + """
+SELECT
+  *
+FROM
+  base_query
+WHERE
+  schema_name='{schema_name}'
+  AND
+  table_view='{schema_name}.{table_name}'
+
+"""
+)
+
 
 class TestIndex:
     @pytest.fixture(scope="class")
@@ -143,3 +170,77 @@ def test_create_index(self, project):
             "Nonclustered unique index": 4,
         }
         assert schema_dict == expected
+
+
+class TestIndexDropsOnlySchema:
+    @pytest.fixture(scope="class")
+    def project_config_update(self):
+        return {"name": "generic_tests"}
+
+    @pytest.fixture(scope="class")
+    def seeds(self):
+        return {
+            "raw_data.csv": index_seed_csv,
+            "schema.yml": index_schema_base_yml,
+        }
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "index_model.sql": drop_schema_model,
+            "index_ccs_model.sql": model_sql_ccs,
+            "schema.yml": model_yml,
+        }
+
+    def create_table_and_index_other_schema(self, project):
+        _schema = project.test_schema + "other"
+        create_sql = f"""
+        USE [{project.database}];
+        IF NOT EXISTS (SELECT * FROM sys.schemas WHERE name = '{_schema}')
+        BEGIN
+        EXEC('CREATE SCHEMA [{_schema}]')
+        END
+        """
+
+        create_table = f"""
+        CREATE TABLE {_schema}.index_model (
+        IDCOL BIGINT
+        )
+        """
+
+        create_index = f"""
+        CREATE INDEX sample_schema ON {_schema}.index_model (IDCOL)
+        """
+        with get_connection(project.adapter):
+            project.adapter.execute(create_sql, fetch=True)
+            project.adapter.execute(create_table)
+            project.adapter.execute(create_index)
+
+    def drop_schema_artifacts(self, project):
+        _schema = project.test_schema + "other"
+        drop_index = f"DROP INDEX IF EXISTS sample_schema ON {_schema}.index_model"
+        drop_table = f"DROP TABLE IF EXISTS {_schema}.index_model"
+        drop_schema = f"DROP SCHEMA IF EXISTS {_schema}"
+
+        with get_connection(project.adapter):
+            project.adapter.execute(drop_index, fetch=True)
+            project.adapter.execute(drop_table)
+            project.adapter.execute(drop_schema)
+
+    def validate_other_schema(self, project):
+        with get_connection(project.adapter):
+            result, table = project.adapter.execute(
+                other_index_count.format(
+                    schema_name=project.test_schema + "other", table_name="index_model"
+                ),
+                fetch=True,
+            )
+
+        assert len(table.rows) == 1
+
+    def test_create_index(self, project):
+        self.create_table_and_index_other_schema(project)
+        run_dbt(["seed"])
+        run_dbt(["run"])
+        self.validate_other_schema(project)
+        self.drop_schema_artifacts(project)

tests/functional/adapter/mssql/test_test_with.py

@@ -0,0 +1,148 @@
+import pytest
+from dbt.tests.util import run_dbt
+
+sample_model = """
+SELECT
+    1 as ID,
+    'a' as data
+
+UNION ALL
+
+SELECT
+    2 as ID,
+    'b' as data
+
+UNION ALL
+
+SELECT
+    2 as ID,
+    'c' as data
+"""
+
+pass_model_yml = """
+version: 2
+models:
+- name: sample_model
+  data_tests:
+  - with_statement_pass:
+      field: ID
+"""
+
+fail_model_yml = """
+version: 2
+models:
+- name: sample_model
+  data_tests:
+  - with_statement_fail:
+      field: ID
+"""
+
+comments_model_yml = """
+version: 2
+models:
+- name: sample_model
+  data_tests:
+  - with_statement_comments:
+      field: ID
+"""
+
+with_test_fail_sql = """
+{% test with_statement_fail(model, field) %}
+
+with test_sample AS (
+    SELECT {{ field }} FROM {{ model }}
+    GROUP BY {{ field }}
+    HAVING COUNT(*) > 1
+)
+SELECT * FROM test_sample
+
+{% endtest %}
+"""
+
+with_test_pass_sql = """
+{% test with_statement_pass(model, field) %}
+
+with test_sample AS (
+    SELECT {{ field }} FROM {{ model }}
+    GROUP BY {{ field }}
+    HAVING COUNT(*) > 2
+)
+SELECT * FROM test_sample
+
+{% endtest %}
+"""
+
+with_test_with_comments_sql = """
+{% test with_statement_comments(model, field) %}
+-- comments
+with test_sample AS (
+    SELECT {{ field }} FROM {{ model }}
+    GROUP BY {{ field }}
+    HAVING COUNT(*) > 2
+)
+SELECT * FROM test_sample
+{% endtest %}
+"""
+
+
+class BaseSQLTestWith:
+    @pytest.fixture(scope="class")
+    def project_config_update(self):
+        return {
+            "config-version": 2,
+            "macro-paths": ["macros"],
+        }
+
+    @pytest.fixture(scope="class")
+    def macros(self):
+        return {
+            "with_statement_pass.sql": with_test_pass_sql,
+            "with_statement_fail.sql": with_test_fail_sql,
+            "with_statement_comments.sql": with_test_with_comments_sql,
+        }
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "sample_model.sql": sample_model,
+            "schema.yml": pass_model_yml,
+        }
+
+
+class TestSQLTestWithPass(BaseSQLTestWith):
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "sample_model.sql": sample_model,
+            "schema.yml": pass_model_yml,
+        }
+
+    def test_sql_test_contains_with(self, project):
+        run_dbt(["run"])
+        run_dbt(["test"])
+
+
+class TestSQLTestWithFail(BaseSQLTestWith):
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "sample_model.sql": sample_model,
+            "schema.yml": fail_model_yml,
+        }
+
+    def test_sql_test_contains_with(self, project):
+        run_dbt(["run"])
+        run_dbt(["test"], expect_pass=False)
+
+
+class TestSQLTestWithComment(BaseSQLTestWith):
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "sample_model.sql": sample_model,
+            "schema.yml": comments_model_yml,
+        }
+
+    def test_sql_test_contains_with(self, project):
+        run_dbt(["run"])
+        run_dbt(["test"])