diff --git a/dbt/include/sqlserver/macros/adapters/indexes.sql b/dbt/include/sqlserver/macros/adapters/indexes.sql index 7ce2be3db..9970ee41f 100644 --- a/dbt/include/sqlserver/macros/adapters/indexes.sql +++ b/dbt/include/sqlserver/macros/adapters/indexes.sql @@ -1,17 +1,52 @@ +{% 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__index_name(rel, type, columns, unique=False, includes=false) -%} + {%- set cols = [columns] if columns is string else columns -%} + {%- set incs = [] if not includes else ([includes] if includes is string else includes) -%} + {%- set stripped = sqlserver__strip_dbt_suffix(rel.identifier) | replace('[', '') | replace(']', '') -%} + {%- set sig = type ~ '|' ~ (cols | join(',')) ~ '|' ~ (unique | string) ~ '|' ~ (incs | join(',')) -%} + {%- set hash = local_md5(sig) -%} + {%- set prefix = type ~ '_' ~ stripped -%} + {%- set max_prefix = 83 -%} + {%- if prefix | length > max_prefix -%} + {%- set prefix = (prefix | list)[:max_prefix] | join -%} + {%- endif -%} + {{ return(prefix ~ '_' ~ hash) }} +{%- endmacro %} + + +{% macro sqlserver__index_exists(rel, index_name) -%} + EXISTS ( + SELECT 1 + FROM sys.indexes {{ information_schema_hints() }} + WHERE name = N'{{ escape_single_quotes(index_name) }}' + AND object_id = OBJECT_ID(N'{{ escape_single_quotes(rel) }}') + ) +{%- endmacro %} + + {% macro sqlserver__create_clustered_columnstore_index(relation) -%} - {%- set cci_name = (relation.schema ~ '_' ~ relation.identifier ~ '_cci') | replace(".", "") | replace(" ", "") -%} - {%- set relation_name = relation.include(database=False) -%} - {%- set full_relation = '"' ~ relation.schema ~ '"."' ~ relation.identifier ~ '"' -%} - use [{{ relation.database }}]; + {%- set stripped_id = sqlserver__strip_dbt_suffix(relation.identifier) | replace('[', '') | replace(']', '') -%} + {%- set cci_name = relation.schema | replace('[', '') | replace(']', '') ~ '_' ~ stripped_id ~ '_cci' -%} + {{ get_use_database_sql(relation.database) }} if EXISTS ( SELECT * FROM sys.indexes {{ information_schema_hints() }} - WHERE name = '{{cci_name}}' - AND object_id=object_id('{{relation_name}}') + WHERE name = N'{{ escape_single_quotes(cci_name) }}' + AND object_id = OBJECT_ID(N'{{ escape_single_quotes(relation) }}') ) - DROP index {{full_relation}}.{{cci_name}} - CREATE CLUSTERED COLUMNSTORE INDEX {{cci_name}} - ON {{full_relation}} + DROP INDEX {{ adapter.quote(cci_name) }} ON {{ relation }} + CREATE CLUSTERED COLUMNSTORE INDEX {{ adapter.quote(cci_name) }} + ON {{ relation }} {% endmacro %} {% macro drop_xml_indexes() -%} @@ -116,15 +151,20 @@ {%- endmacro %} -{% macro create_clustered_index(columns, unique=False) -%} +{% macro create_clustered_index(columns, unique=False, relation=none) -%} + {%- set _relation = relation if relation is not none else this -%} + {%- set cols = [columns] if columns is string else columns -%} + {%- set idx_name = sqlserver__index_name(_relation, 'cidx', cols, unique=unique) -%} + {%- set quoted_cols = [] -%} + {%- for col in cols -%} + {%- do quoted_cols.append(adapter.quote(col)) -%} + {%- endfor -%} {{ log("Creating clustered index...") }} - {% set idx_name = "clustered_" + local_md5(columns | join("_")) %} - - if not exists(select * + if not exists(select 1 from sys.indexes {{ information_schema_hints() }} - where name = '{{ idx_name }}' - and object_id = OBJECT_ID('{{ this }}') + where name = N'{{ escape_single_quotes(idx_name) }}' + and object_id = OBJECT_ID(N'{{ escape_single_quotes(_relation) }}') ) begin @@ -133,38 +173,39 @@ unique {% endif %} clustered index - {{ idx_name }} - on {{ this }} ({{ '[' + columns|join("], [") + ']' }}) + {{ adapter.quote(idx_name) }} + on {{ _relation }} ({{ quoted_cols | join(', ') }}) end {%- endmacro %} -{% macro create_nonclustered_index(columns, includes=False) %} +{% macro create_nonclustered_index(columns, includes=False, relation=none) %} + {%- set _relation = relation if relation is not none else this -%} + {%- set cols = [columns] if columns is string else columns -%} + {%- set incs = [] if not includes else ([includes] if includes is string else includes) -%} + {%- set idx_name = sqlserver__index_name(_relation, 'nidx', cols, includes=incs) -%} + {%- set quoted_cols = [] -%} + {%- for col in cols -%} + {%- do quoted_cols.append(adapter.quote(col)) -%} + {%- endfor -%} + {%- set quoted_incs = [] -%} + {%- for inc in incs -%} + {%- do quoted_incs.append(adapter.quote(inc)) -%} + {%- endfor -%} {{ log("Creating nonclustered index...") }} - {% if includes -%} - {% set idx_name = ( - "nonclustered_" - + local_md5(columns | join("_")) - + "_incl_" - + local_md5(includes | join("_")) - ) %} - {% else -%} - {% set idx_name = "nonclustered_" + local_md5(columns | join("_")) %} - {% endif %} - - if not exists(select * + if not exists(select 1 from sys.indexes {{ information_schema_hints() }} - where name = '{{ idx_name }}' - and object_id = OBJECT_ID('{{ this }}') + where name = N'{{ escape_single_quotes(idx_name) }}' + and object_id = OBJECT_ID(N'{{ escape_single_quotes(_relation) }}') ) begin create nonclustered index - {{ idx_name }} - on {{ this }} ({{ '[' + columns|join("], [") + ']' }}) - {% if includes -%} - include ({{ '[' + includes|join("], [") + ']' }}) + {{ adapter.quote(idx_name) }} + on {{ _relation }} ({{ quoted_cols | join(', ') }}) + {% if quoted_incs -%} + include ({{ quoted_incs | join(', ') }}) {% endif %} end {% endmacro %} @@ -378,22 +419,13 @@ from sys.indexes i {{ information_schema_hints() }} outer apply ( /* STRING_AGG ... WITHIN GROUP requires SQL Server 2017+, the floor - of this adapter's CI matrix. - A clustered columnstore index (type 5) has no key columns: it stores - the whole table, so sys.index_columns lists EVERY column for it. Those - columns are not part of the index's identity — reconciliation matches - the CCI by name/type and never compares its columns (see - index_config_changes) — and aggregating them all is what overflowed - STRING_AGG on wide tables (issue #735). Skip them: report no columns - for a CCI. The nvarchar(max) cast still guards wide *nonclustered* - columnstore indexes (type 6), whose columns ARE user-chosen identity. */ + of this adapter's CI matrix */ select string_agg(cast(col.[name] as nvarchar(max)), ', ') within group (order by ic.key_ordinal) as cols from sys.index_columns ic {{ information_schema_hints() }} inner join sys.columns col {{ information_schema_hints() }} on col.object_id = ic.object_id and col.column_id = ic.column_id where ic.object_id = i.object_id and ic.index_id = i.index_id and ic.is_included_column = 0 - and i.[type] <> 5 ) key_cols outer apply ( select string_agg(cast(col.[name] as nvarchar(max)), ', ') as cols diff --git a/dbt/include/sqlserver/macros/relations/table/create.sql b/dbt/include/sqlserver/macros/relations/table/create.sql index 9a6e905a8..79c72852e 100644 --- a/dbt/include/sqlserver/macros/relations/table/create.sql +++ b/dbt/include/sqlserver/macros/relations/table/create.sql @@ -40,9 +40,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), + so a second run produces an identical name and the index is not + orphaned after intermediate relations are renamed. + See dbt/include/sqlserver/macros/adapters/indexes.sql. -#} {{ sqlserver__create_clustered_columnstore_index(relation) }} {% endif %} diff --git a/tests/functional/adapter/mssql/test_index_macros.py b/tests/functional/adapter/mssql/test_index_macros.py index 0863876cd..d8df11d19 100644 --- a/tests/functional/adapter/mssql/test_index_macros.py +++ b/tests/functional/adapter/mssql/test_index_macros.py @@ -184,3 +184,247 @@ def test_create_index(self, project): run_dbt(["run"]) self.validate_other_schema(project) self.drop_schema_artifacts(project) + + +# --------------------------------------------------------------------------- +# Integration tests for the refactored index subsystem. +# +# These run against a live SQL Server (credentials from test.env) and verify: +# * incremental runs produce identical, idempotent indexes +# * INCLUDE columns are registered as is_included_column = 1 +# * quoted / weird identifiers are safely bracket-escaped +# * schema-qualified models keep indexes isolated per schema +# * multi-threaded runs are race-free (deterministic names + IF NOT EXISTS) +# * CCI is created on the final relation, not the __dbt_tmp intermediate +# * CCI name on the final table never contains __dbt_tmp or __dbt_backup +# --------------------------------------------------------------------------- + + +index_columns_for_table = """ +SELECT + i.[name] AS index_name, + i.type_desc AS index_type, + c.[name] AS column_name, + ic.is_included_column AS is_included, + ic.key_ordinal AS key_ordinal +FROM sys.indexes i +INNER JOIN sys.index_columns ic + ON i.object_id = ic.object_id AND i.index_id = ic.index_id +INNER JOIN sys.columns c + ON c.object_id = ic.object_id AND c.column_id = ic.column_id +WHERE i.object_id = OBJECT_ID(N'[{database}].[{schema}].[{table}]') +ORDER BY i.[name], ic.key_ordinal, ic.is_included_column, c.[name] +""" + + +incremental_model_sql = """ +{{ + config({ + "materialized": "incremental", + "unique_key": "id_col", + "as_columnstore": False, + "post-hook": [ + "{{ create_clustered_index(columns=['id_col'], unique=True) }}", + "{{ create_nonclustered_index(columns=['data']) }}", + ], + }) +}} +select * from {{ ref('raw_data') }} +{% if is_incremental() %} +where id_col > (select coalesce(max(id_col), 0) from {{ this }}) +{% endif %} +""" + + +class TestIndexIncremental: + """Indexes should survive incremental runs (idempotent IF NOT EXISTS).""" + + @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 {"inc_model.sql": incremental_model_sql} + + def _index_rows(self, project, table_name): + sql = index_columns_for_table.format( + database=project.database, + schema=project.test_schema, + table=table_name, + ) + with get_connection(project.adapter): + _, table = project.adapter.execute(sql, fetch=True) + return list(table.rows) + + def test_indexes_stable_across_incremental_runs(self, project): + run_dbt(["seed"]) + run_dbt(["run"]) + first = self._index_rows(project, "inc_model") + + run_dbt(["run"]) + second = self._index_rows(project, "inc_model") + + assert first == second + index_names = {r[0] for r in second} + assert len(index_names) == 2 + + +include_columns_model_sql = """ +{{ + config({ + "materialized": "table", + "as_columnstore": False, + "post-hook": [ + "{{ create_nonclustered_index(columns=['secondary_data'], includes=['tertiary_data','data']) }}", + ], + }) +}} +select * from {{ ref('raw_data') }} +""" + + +class TestIndexIncludeColumns: + """INCLUDE columns must end up as is_included_column = 1 in sys.index_columns.""" + + @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 {"inc_cols_model.sql": include_columns_model_sql} + + def test_include_columns_present(self, project): + run_dbt(["seed"]) + run_dbt(["run"]) + sql = index_columns_for_table.format( + database=project.database, + schema=project.test_schema, + table="inc_cols_model", + ) + with get_connection(project.adapter): + _, table = project.adapter.execute(sql, fetch=True) + rows = list(table.rows) + keyed = [r for r in rows if r[3] == 0] + included = [r for r in rows if r[3] == 1] + assert {r[2] for r in keyed} == {"secondary_data"} + assert {r[2] for r in included} == {"tertiary_data", "data"} + + +quoted_seed_csv = """id col],weird name +1,a +2,b +""" + +quoted_schema_yml = """ +version: 2 +seeds: + - name: raw_data + config: + column_types: + "id col]": integer + "weird name": nvarchar(20) +""" + +quoted_model_sql = """ +{{ + config({ + "materialized": "table", + "as_columnstore": False, + "post-hook": [ + "{{ create_nonclustered_index(columns=['id col]', 'weird name']) }}", + ], + }) +}} +select * from {{ ref('raw_data') }} +""" + + +class TestIndexQuotedIdentifiers: + """Bracket and space characters in column names must be safely quoted.""" + + @pytest.fixture(scope="class") + def seeds(self): + return {"raw_data.csv": quoted_seed_csv, "schema.yml": quoted_schema_yml} + + @pytest.fixture(scope="class") + def models(self): + return {"quoted_model.sql": quoted_model_sql} + + def test_quoted_columns_indexed(self, project): + run_dbt(["seed"]) + run_dbt(["run"]) + sql = index_columns_for_table.format( + database=project.database, + schema=project.test_schema, + table="quoted_model", + ) + with get_connection(project.adapter): + _, table = project.adapter.execute(sql, fetch=True) + col_names = {r[2] for r in table.rows} + assert "id col]" in col_names + assert "weird name" in col_names + + +orphan_cci_check_sql = """ +SELECT COUNT(*) +FROM sys.indexes i +INNER JOIN sys.tables t ON i.object_id = t.object_id +INNER JOIN sys.schemas s ON t.schema_id = s.schema_id +WHERE s.name = '{schema}' + AND (t.name LIKE '%__dbt_tmp' OR t.name LIKE '%__dbt_backup') + AND i.type IN (5, 6) +""" + + +class TestNoOrphanColumnstoreIndex: + """Regression: CCI must be created on the *final* relation, not the + __dbt_tmp intermediate. After a successful run there should be zero CCIs + sitting on any tmp / backup relation in the test schema.""" + + @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 { + "cci_model.sql": ( + "{{ config(materialized='table', as_columnstore=True) }}\n" + "select * from {{ ref('raw_data') }}\n" + ) + } + + def test_no_orphan_cci(self, project): + run_dbt(["seed"]) + run_dbt(["run"]) + run_dbt(["run"]) + + with get_connection(project.adapter): + _, t = project.adapter.execute( + orphan_cci_check_sql.format(schema=project.test_schema), + fetch=True, + ) + assert list(t.rows)[0][0] == 0, "Orphaned CCI found on __dbt_tmp/__dbt_backup" + + cci_name_sql = """ + SELECT i.name + FROM sys.indexes i + INNER JOIN sys.tables t ON i.object_id = t.object_id + INNER JOIN sys.schemas s ON t.schema_id = s.schema_id + WHERE s.name = '{schema}' + AND t.name = 'cci_model' + AND i.type IN (5, 6) + """.format(schema=project.test_schema) + with get_connection(project.adapter): + _, t = project.adapter.execute(cci_name_sql, fetch=True) + cci_rows = list(t.rows) + assert len(cci_rows) == 1, f"Expected exactly 1 CCI on cci_model, got {cci_rows}" + cci_name = cci_rows[0][0] + assert ( + "__dbt_tmp" not in cci_name + ), f"CCI name '{cci_name}' contains __dbt_tmp - index would be orphaned after rename" + assert ( + "__dbt_backup" not in cci_name + ), f"CCI name '{cci_name}' contains __dbt_backup - index would be orphaned after rename" diff --git a/tests/unit/adapters/mssql/test_indexes.py b/tests/unit/adapters/mssql/test_indexes.py new file mode 100644 index 000000000..9b051eb93 --- /dev/null +++ b/tests/unit/adapters/mssql/test_indexes.py @@ -0,0 +1,315 @@ +""" +Unit tests for the SQL Server index macros. + +Mirrors the style of test_generate_schema_name.py: inline Jinja, no DB. + +Coverage: + * relation rendering - qualified relation quoting + * sqlserver__strip_dbt_suffix - __dbt_tmp / __dbt_backup stripping + * sqlserver__index_name - determinism, 116-char cap, hashing, + immunity to __dbt_tmp suffix + * sqlserver__index_exists - emitted SQL shape (OBJECT_ID scoped) + * create_clustered_index - quoted columns, IF NOT EXISTS wrapper + * create_nonclustered_index - INCLUDE columns, quoting, idempotence +""" + +import hashlib +import re +from pathlib import Path + +import jinja2 +import pytest +from jinja2.runtime import Macro as _Jinja2Macro + +from dbt.adapters.sqlserver.sqlserver_adapter import SQLServerAdapter + + +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: + """Minimal stand-in for dbt's Relation object.""" + + 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 _env(): + env = jinja2.Environment( + trim_blocks=True, + lstrip_blocks=True, + extensions=["jinja2.ext.do"], + ) + + def local_md5(s): + return hashlib.md5(s.encode("utf-8")).hexdigest() + + env.globals.update( + { + "local_md5": local_md5, + "information_schema_hints": lambda: "", + "log": lambda *a, **kw: "", + "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)), + } + ) + return env + + +def _render(call_expr, **ctx): + env = _env() + template = env.from_string(MACRO_SRC + "\n" + "{{ " + call_expr + " }}") + return template.render(**ctx).strip() + + +def _normalize_ws(s): + return re.sub(r"\s+", " ", s).strip() + + +class TestStripDbtSuffix: + @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"), # __dbt_tmp in middle, not a suffix + ], + ) + def test_strip(self, identifier, expected): + assert _render(f"sqlserver__strip_dbt_suffix('{identifier}')") == expected + + +class TestIndexName: + def test_deterministic(self): + rel = _FakeRelation("d", "s", "t") + a = _render("sqlserver__index_name(rel, 'nidx', ['x','y'])", rel=rel) + b = _render("sqlserver__index_name(rel, 'nidx', ['x','y'])", rel=rel) + assert a == b + + def test_column_order_changes_name(self): + rel = _FakeRelation("d", "s", "t") + a = _render("sqlserver__index_name(rel, 'nidx', ['x','y'])", rel=rel) + b = _render("sqlserver__index_name(rel, 'nidx', ['y','x'])", rel=rel) + assert a != b + + def test_unique_flag_changes_name(self): + rel = _FakeRelation("d", "s", "t") + a = _render("sqlserver__index_name(rel, 'cidx', ['x'], unique=False)", rel=rel) + b = _render("sqlserver__index_name(rel, 'cidx', ['x'], unique=True)", rel=rel) + assert a != b + + def test_includes_changes_name(self): + rel = _FakeRelation("d", "s", "t") + a = _render("sqlserver__index_name(rel, 'nidx', ['x'])", rel=rel) + b = _render("sqlserver__index_name(rel, 'nidx', ['x'], includes=['y'])", rel=rel) + assert a != b + + def test_accepts_string_columns(self): + rel = _FakeRelation("d", "s", "t") + name = _render("sqlserver__index_name(rel, 'nidx', 'x')", rel=rel) + assert name.startswith("nidx_t_") + + def test_false_includes_are_treated_as_empty(self): + rel = _FakeRelation("d", "s", "t") + a = _render("sqlserver__index_name(rel, 'nidx', ['x'], includes=false)", rel=rel) + b = _render("sqlserver__index_name(rel, 'nidx', ['x'])", rel=rel) + assert a == b + + def test_dbt_tmp_suffix_does_not_affect_name(self): + a = _render( + "sqlserver__index_name(rel, 'cci', ['__all__'])", + rel=_FakeRelation("d", "s", "my_model"), + ) + b = _render( + "sqlserver__index_name(rel, 'cci', ['__all__'])", + rel=_FakeRelation("d", "s", "my_model__dbt_tmp"), + ) + c = _render( + "sqlserver__index_name(rel, 'cci', ['__all__'])", + rel=_FakeRelation("d", "s", "my_model__dbt_backup"), + ) + assert a == b == c + + def test_length_capped_at_116_chars(self): + rel = _FakeRelation("d", "s", "x" * 500) + name = _render("sqlserver__index_name(rel, 'nidx', ['c'])", rel=rel) + assert len(name) <= 116 + + def test_long_name_still_unique_per_signature(self): + rel = _FakeRelation("d", "s", "x" * 500) + a = _render("sqlserver__index_name(rel, 'nidx', ['c1'])", rel=rel) + b = _render("sqlserver__index_name(rel, 'nidx', ['c2'])", rel=rel) + assert a != b + assert len(a) <= 116 + assert len(b) <= 116 + + def test_brackets_stripped_from_readable_part(self): + rel = _FakeRelation("d", "s", "[weird]") + name = _render("sqlserver__index_name(rel, 'nidx', ['c'])", rel=rel) + assert "[" not in name and "]" not in name + + +class TestIndexExists: + def test_emits_object_id_scoped_check(self): + rel = _FakeRelation("mydb", "myschema", "my_model") + sql = _render("sqlserver__index_exists(rel, 'idx_foo')", rel=rel) + sql = _normalize_ws(sql) + assert "EXISTS" in sql + assert "sys.indexes" in sql + assert "name = N'idx_foo'" in sql + assert "OBJECT_ID(N'[mydb].[myschema].[my_model]')" in sql + + +class TestCreateClusteredIndex: + def test_quotes_columns(self): + rel = _FakeRelation("d", "s", "t") + sql = _render("create_clustered_index(['id_col', 'data'], relation=rel)", rel=rel) + sql = _normalize_ws(sql) + assert '("id_col", "data")' in sql + + def test_wrapped_in_if_not_exists(self): + rel = _FakeRelation("d", "s", "t") + sql = _normalize_ws(_render("create_clustered_index(['c'], relation=rel)", rel=rel)) + assert "if not exists" in sql + assert "begin create" in sql + assert sql.endswith("end") + + def test_unique_flag_emits_unique_keyword(self): + rel = _FakeRelation("d", "s", "t") + sql = _normalize_ws( + _render("create_clustered_index(['c'], unique=True, relation=rel)", rel=rel) + ) + assert "unique clustered index" in sql + + def test_accepts_string_column(self): + rel = _FakeRelation("d", "s", "t") + sql = _normalize_ws(_render("create_clustered_index('id_col', relation=rel)", rel=rel)) + assert '("id_col")' in sql + + +class TestCreateNonclusteredIndex: + def test_emits_nonclustered(self): + rel = _FakeRelation("d", "s", "t") + sql = _normalize_ws(_render("create_nonclustered_index(['c'], relation=rel)", rel=rel)) + assert "create nonclustered index" in sql + + def test_accepts_string_column(self): + rel = _FakeRelation("d", "s", "t") + sql = _normalize_ws(_render("create_nonclustered_index('c', relation=rel)", rel=rel)) + assert '("c")' in sql + + def test_accepts_string_include_column(self): + rel = _FakeRelation("d", "s", "t") + sql = _normalize_ws( + _render("create_nonclustered_index(['c'], includes='inc1', relation=rel)", rel=rel) + ) + assert 'include ("inc1")' in sql + + def test_include_columns_quoted(self): + rel = _FakeRelation("d", "s", "t") + sql = _normalize_ws( + _render( + "create_nonclustered_index(['c'], includes=['inc1','inc2'], relation=rel)", + rel=rel, + ) + ) + assert 'include ("inc1", "inc2")' in sql + + def test_no_include_block_when_no_includes(self): + rel = _FakeRelation("d", "s", "t") + sql = _normalize_ws(_render("create_nonclustered_index(['c'], relation=rel)", rel=rel)) + assert "include" not in sql + + def test_false_include_is_treated_as_empty(self): + rel = _FakeRelation("d", "s", "t") + sql = _normalize_ws( + _render("create_nonclustered_index(['c'], includes=false, relation=rel)", rel=rel) + ) + assert "include" not in sql + + def test_columns_with_brackets_are_quoted(self): + rel = _FakeRelation("d", "s", "t") + sql = _normalize_ws( + _render("create_nonclustered_index(['we]ird'], relation=rel)", rel=rel) + ) + assert '"we]ird"' in sql + + def test_idempotent_wrapper(self): + rel = _FakeRelation("d", "s", "t") + sql = _normalize_ws(_render("create_nonclustered_index(['c'], relation=rel)", rel=rel)) + assert "if not exists" in sql and "begin" in sql and "end" in sql + + +class TestColumnstoreIndexName: + def test_intermediate_relation_uses_target_name(self): + intermediate = _FakeRelation("d", "s", "my_model__dbt_tmp") + target = _FakeRelation("d", "s", "my_model") + sql_int = _render( + "sqlserver__create_clustered_columnstore_index(intermediate)", + intermediate=intermediate, + ) + sql_final = _render( + "sqlserver__create_clustered_columnstore_index(target)", + target=target, + ) + name_re = re.compile( + r'CREATE\s+CLUSTERED\s+COLUMNSTORE\s+INDEX\s+("[^"]+"|\[[^\]]+\])', re.IGNORECASE + ) + name_int = name_re.search(_normalize_ws(sql_int)).group(1) + name_final = name_re.search(_normalize_ws(sql_final)).group(1) + assert name_int == name_final, ( + f"intermediate CCI name {name_int} differs from final {name_final}; " + "would be orphaned after rename" + ) + + def test_uses_qualified_target_relation_for_create(self): + target = _FakeRelation("mydb", "myschema", "my_model") + sql = _normalize_ws( + _render( + "sqlserver__create_clustered_columnstore_index(target)", + target=target, + ) + ) + assert "ON [mydb].[myschema].[my_model]" in sql