fix(indexes): avoid __dbt_tmp/__dbt_backup leakage in CCI names (#578)

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

@@ -8,10 +8,22 @@
 {%- endmacro %}
 
 
+{% macro sqlserver__strip_dbt_suffix(identifier) -%}
+    {%- set ns = namespace(result=identifier) -%}
+    {%- for suffix in ['__dbt_tmp_vw', '__dbt_backup', '__dbt_tmp'] -%}
+        {%- if ns.result.endswith(suffix) -%}
+            {%- set ns.result = ns.result[:(ns.result | length) - (suffix | length)] -%}
+        {%- endif -%}
+    {%- endfor -%}
+    {{ return(ns.result) }}
+{%- endmacro %}
+
+
 {% macro sqlserver__create_clustered_columnstore_index(relation) -%}
     {#- cci_name embeds the schema, so it must be quoted as an identifier
         (raw only in the string comparison below) -- issue #409 -#}
-    {%- set cci_name = (relation.schema ~ '_' ~ relation.identifier ~ '_cci') | replace(".", "") | replace(" ", "") -%}
+    {%- set stripped_identifier = sqlserver__strip_dbt_suffix(relation.identifier) -%}
+    {%- set cci_name = (relation.schema ~ '_' ~ stripped_identifier ~ '_cci') | replace(".", "") | replace(" ", "") -%}
     {%- set relation_name = relation.include(database=False) -%}
     {{ get_use_database_sql(relation.database) }}
     if EXISTS (

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

@@ -64,9 +64,11 @@
     {% set as_columnstore = config.get('as_columnstore', default=true) %}
     {% if not temporary and as_columnstore -%}
         {#-
-        add columnstore index
-        this creates with dbt_temp as its coming from a temporary relation before renaming
-        could alter relation to drop the dbt_temp portion if needed
+        Add a clustered columnstore index. The index name is derived from the
+        *final* relation name (with __dbt_tmp / __dbt_backup suffixes stripped),
+        This keeps generated names aligned with final-relation naming and
+        avoids suffix leakage from intermediate relation identifiers.
+        See dbt/include/sqlserver/macros/adapters/indexes.sql.
         -#}
         {{ sqlserver__create_clustered_columnstore_index(relation) }}
    {% endif %}

tests/unit/adapters/mssql/test_index_configs.py

@@ -1,16 +1,118 @@
+from pathlib import Path
 from unittest.mock import MagicMock
 
+import jinja2
 import pytest
+from jinja2.runtime import Macro as _Jinja2Macro
 
 from dbt.adapters.exceptions import IndexConfigError, IndexConfigNotDictError
 from dbt.adapters.sqlserver.relation_configs.index import SQLServerIndexConfig, SQLServerIndexType
+from dbt.adapters.sqlserver.sqlserver_adapter import SQLServerAdapter
 from dbt.exceptions import DbtRuntimeError
 
 
 def test_sqlserver_index_type_default():
     assert SQLServerIndexType.default() == SQLServerIndexType.nonclustered
 
 
+class _MacroReturn(BaseException):
+    def __init__(self, value):
+        self.value = value
+
+
+_orig_macro_call = _Jinja2Macro.__call__
+
+
+def _patched_macro_call(self, *args, **kwargs):
+    try:
+        return _orig_macro_call(self, *args, **kwargs)
+    except _MacroReturn as exc:
+        return exc.value
+
+
+@pytest.fixture(scope="module", autouse=True)
+def _patch_jinja2_macro_return():
+    _Jinja2Macro.__call__ = _patched_macro_call
+    yield
+    _Jinja2Macro.__call__ = _orig_macro_call
+
+
+MACRO_PATH = (
+    Path(__file__).resolve().parents[4]
+    / "dbt"
+    / "include"
+    / "sqlserver"
+    / "macros"
+    / "adapters"
+    / "indexes.sql"
+)
+
+MACRO_SRC = MACRO_PATH.read_text(encoding="utf-8")
+
+
+class _FakeRelation:
+    def __init__(self, database, schema, identifier):
+        self.database = database
+        self.schema = schema
+        self.identifier = identifier
+
+    def __str__(self):
+        return f"[{self.database}].[{self.schema}].[{self.identifier}]"
+
+    def include(self, database=False):
+        if database:
+            return str(self)
+        return f"[{self.schema}].[{self.identifier}]"
+
+
+def _render_index_macro(call_expr, **ctx):
+    env = jinja2.Environment(
+        trim_blocks=True,
+        lstrip_blocks=True,
+        extensions=["jinja2.ext.do"],
+    )
+    env.globals.update(
+        {
+            "information_schema_hints": lambda: "",
+            "get_use_database_sql": lambda db: f"USE [{db}];",
+            "escape_single_quotes": lambda value: str(value).replace("'", "''"),
+            "adapter": SQLServerAdapter,
+            "this": _FakeRelation("mydb", "myschema", "my_model"),
+            "return": lambda v: (_ for _ in ()).throw(_MacroReturn(v)),
+        }
+    )
+
+    template = env.from_string(MACRO_SRC + "\n" + "{{ " + call_expr + " }}")
+    return template.render(**ctx).strip()
+
+
+class TestIssue578IndexNaming:
+    @pytest.mark.parametrize(
+        "identifier, expected",
+        [
+            ("my_model", "my_model"),
+            ("my_model__dbt_tmp", "my_model"),
+            ("my_model__dbt_backup", "my_model"),
+            ("my_model__dbt_tmp_vw", "my_model"),
+            ("my__dbt_tmp_model", "my__dbt_tmp_model"),
+        ],
+    )
+    def test_strip_dbt_suffix(self, identifier, expected):
+        assert _render_index_macro(f"sqlserver__strip_dbt_suffix('{identifier}')") == expected
+
+    def test_generated_cci_name_removes_tmp_suffix(self):
+        rel = _FakeRelation("mydb", "myschema", "my_model__dbt_tmp")
+        sql = _render_index_macro("sqlserver__create_clustered_columnstore_index(rel)", rel=rel)
+        assert "my_model__dbt_tmp_cci" not in sql
+        assert "my_model_cci" in sql
+
+    def test_generated_cci_name_removes_backup_suffix(self):
+        rel = _FakeRelation("mydb", "myschema", "my_model__dbt_backup")
+        sql = _render_index_macro("sqlserver__create_clustered_columnstore_index(rel)", rel=rel)
+        assert "my_model__dbt_backup_cci" not in sql
+        assert "my_model_cci" in sql
+
+
 def test_sqlserver_index_type_valid_types():
     valid_types = SQLServerIndexType.valid_types()
     assert isinstance(valid_types, tuple)