Add schema filter to index drop query

Author: cody-scott

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;

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)