fix: rebuild the DML refresh's scratch table so a schema change keeps the columnstore

CHANGELOG.md

@@ -21,6 +21,7 @@
 - Fix models failing with `Incorrect syntax near '\'` when the schema name needs delimiters, such as a domain-qualified `domain\user`. The clustered columnstore index name embeds the schema and was emitted as a bare identifier, so the generated DDL did not parse. [#409](https://github.com/dbt-msft/dbt-sqlserver/issues/409)
 - Fix identifiers built inside string literals not being quoted, which broke schema names containing a `.` or a `"`. `OBJECT_ID('schema.table')` returns `NULL` rather than erroring for such a name, so the failures were silent: the drop-before-create guards in `create_table_as` treated an existing table as absent (then hit `Msg 2714`), and the mask introspection in `apply_masks` found no columns, so configured masks were never applied. `sp_rename` was affected too, failing the table rename-swap with `No item by the name of ...`. All now pass quoted, qualified names. [#785](https://github.com/dbt-msft/dbt-sqlserver/issues/785)
 - Fix the `sqlserver__openquery` macro to quote linked-server names through `adapter.quote()`, keeping its generated identifier style consistent with the rest of the adapter.
+- Fix a model with `table_refresh_method: dml` losing its clustered columnstore index - and, under an enforced contract, its `NOT NULL`s as well - the first time its schema changed. That path builds a scratch table and, when the columns no longer match, renames it into position; the scratch was built with `SELECT * INTO`, which copies no index and no constraint and infers nullability from the query rather than from the contract, and `create_indexes` only builds what the `indexes` config names, never the `as_columnstore` CCI. So the model came back stripped and stayed that way, since every later run matched the new schema and took the DELETE+INSERT path. That branch now rebuilds the scratch table through `create_table_as` - the way every other build path in the adapter creates a table - which carries the full column DDL and the columnstore index across the swap. On a schema-change run - and only there - that costs a second execution of the model's SQL (once for the `SELECT ... INTO` schema probe, once for the rebuild) and one extra columnstore build. The rebuild bulk-loads the same way the model would on any other build path - `SELECT ... INTO` for an ordinary model, `CREATE TABLE` plus `INSERT ... WITH (TABLOCK)` under an enforced contract - so it gives up no minimal logging to do it. Steady-state refreshes are unchanged. A table already stripped by this bug is not repaired by upgrading: its schema still matches, so it stays on the DELETE+INSERT path, and index reconciliation protects an existing columnstore index without ever creating a missing one. To rebuild it, temporarily set `full_refresh_build: prebuilt` and run with `--full-refresh`.
 
 #### Under the hood
 

README.md

@@ -267,6 +267,8 @@ You can also set it per model:
 {{ config(materialized="table", as_columnstore=false) }}
 ```
 
+With `table_refresh_method: dml`, a schema change makes the refresh fall back to a rename-swap. On that run — and only that run — the scratch table is rebuilt the way this adapter builds every other table, so it carries the model's columnstore index, and under an enforced contract its `NOT NULL`s, into the swap. That run therefore executes the model's SQL twice — once for the `SELECT … INTO` that probes for the schema change, once for the rebuild — and builds the columnstore index once. Steady-state refreshes are unaffected and keep the single `SELECT … INTO`. A table that lost its columnstore index to this bug before you upgraded is not repaired automatically: its schema still matches, so it stays on the cheap path. To rebuild it, temporarily set `full_refresh_build: prebuilt` and run with `--full-refresh`.
+
 ### Dynamic Data Masking (`masked_with` / `masks`)
 
 The adapter can apply SQL Server [Dynamic Data Masking](https://learn.microsoft.com/en-us/sql/relational-databases/security/dynamic-data-masking) (DDM) to columns as part of the materialization, so masks are re-applied on every build and survive dbt's drop-and-recreate on a full refresh. A principal granted `SELECT` but not `UNMASK` then sees masked values instead of real data (dbt's own build principal, being `db_owner`, keeps `UNMASK` and reads real data). Requires **SQL Server 2016+**.

dbt/include/sqlserver/macros/materializations/models/table/table_dml_refresh.sql

@@ -110,6 +110,40 @@
     {%- set backup_relation = make_backup_relation(target_relation, backup_relation_type) -%}
     {{ drop_relation_if_exists(backup_relation) }}
 
+    {#- The scratch table above came from SELECT * INTO, which is the right
+        shape for the schema probe and the wrong one for the object that is
+        about to be renamed into position: it copies no constraint and no
+        index, and takes nullability from the query rather than from a
+        contract. Left as-is it silently strips the model of its clustered
+        columnstore index - create_indexes only builds what the `indexes`
+        config names, never the as_columnstore CCI - and, under a contract, of
+        its NOT NULLs and inline constraints too. None of it came back on a
+        later run, because every later run matched the new schema and took the
+        DELETE+INSERT path above.
+
+        So rebuild it the way this adapter builds every other table. The
+        rebuild belongs on this branch alone: doing it up front would build,
+        and then throw away, a full columnstore index on every steady-state
+        refresh, which on a large table dominates the run. A schema change is
+        rare, so one extra build here is much the cheaper trade.
+
+        It is not free, though: the model's SQL runs a second time here, the
+        SELECT * INTO above having already run it once as the schema probe.
+        Any side effect in that SQL therefore happens twice, and the two runs
+        are not interchangeable - the schema decision came from the first, the
+        table renamed into position comes from the second. A model whose column
+        shape can differ between them lands the second shape unchecked, unless
+        a contract is enforced and create_table_as re-asserts it. Probing the
+        tmp view instead of the materialized scratch would collapse the two
+        back into one, at the cost of changing how the probe behaves - a
+        separate change, not this one.
+
+        create_table_as builds and drops its own __dbt_tmp_vw. -#}
+    {{ drop_relation_if_exists(refresh_relation) }}
+    {% call statement('dml_refresh_rebuild') -%}
+      {{ get_create_table_as_sql(False, refresh_relation, sql) }}
+    {%- endcall %}
+
     {# Rename scratch table into position #}
     {% set existing_relation = load_cached_relation(target_relation) %}
     {% if existing_relation is not none %}
@@ -118,7 +152,7 @@
 
     {{ adapter.rename_relation(refresh_relation, target_relation) }}
 
-    {# Rebuilt via SELECT INTO (no masks carried), so apply masks before
+    {# Freshly rebuilt (no masks carried), so apply masks before
        create_indexes — a mask cannot be added to a column an index depends
        on (documented for all SQL Server versions). #}
     {% do apply_masks(target_relation, mask_config) %}

tests/functional/adapter/mssql/test_openquery.py

@@ -21,7 +21,27 @@
 
 from dbt.tests.util import run_dbt
 
-_LINKED_SERVER_NAME = "LOCALLOOP"
+# A linked server is instance-wide, so a fixed name collides when pytest-xdist
+# runs this class on more than one worker against the same SQL Server: the
+# setup below drops-and-recreates, and the teardown drops outright, so one
+# worker pulls the server out from under another mid-run. The victim's models
+# then fail with Msg 7202 ("Could not find server ... in sys.servers") even
+# though its own fixture verified the server existed moments earlier - the
+# creator sees its own row, the other worker's connection does not.
+#
+# One name per worker process keeps them apart. It is stable across reruns
+# (unlike a uuid), so the IF EXISTS guard below still cleans up a server left
+# behind by a crashed run rather than leaking a new one each time.
+_PLACEHOLDER_SERVER_NAME = "LOCALLOOP"
+_LINKED_SERVER_NAME = "{}_{}".format(
+    _PLACEHOLDER_SERVER_NAME,
+    os.environ.get("PYTEST_XDIST_WORKER", "main").upper(),
+)
+
+
+def _for_this_worker(sql: str) -> str:
+    """Point SQL written against the placeholder name at this worker's server."""
+    return sql.replace(_PLACEHOLDER_SERVER_NAME, _LINKED_SERVER_NAME)
 
 
 def _create_linked_server_sql(major_version: int) -> str:
@@ -161,14 +181,17 @@ class TestOpenquery:
     @pytest.fixture(scope="class")
     def models(self):
         return {
-            "basic_model.sql": basic_sql,
-            "quotes_model.sql": quotes_sql,
-            "cr_model.sql": cr_sql,
-            "max_length_model.sql": max_length_sql,
-            "empty_server_model.sql": empty_server_sql,
-            "none_server_model.sql": none_server_sql,
-            "empty_remote_model.sql": empty_remote_sql,
-            "too_long_model.sql": too_long_sql,
+            name: _for_this_worker(body)
+            for name, body in {
+                "basic_model.sql": basic_sql,
+                "quotes_model.sql": quotes_sql,
+                "cr_model.sql": cr_sql,
+                "max_length_model.sql": max_length_sql,
+                "empty_server_model.sql": empty_server_sql,
+                "none_server_model.sql": none_server_sql,
+                "empty_remote_model.sql": empty_remote_sql,
+                "too_long_model.sql": too_long_sql,
+            }.items()
         }
 
     @pytest.fixture(scope="class")
@@ -205,7 +228,7 @@ def _run_all(self, project, _linked_server):
     def test_emits_openquery_and_returns_rows(self, project, _run_all):
         """Happy path: quoted server name, literal remote SQL, real rows."""
         sql = _find_compiled_sql(project, "basic_model.sql")
-        assert 'OPENQUERY("LOCALLOOP", \'SELECT 1 AS id' in sql
+        assert _for_this_worker('OPENQUERY("LOCALLOOP", \'SELECT 1 AS id') in sql
         rows = project.run_sql(
             f"SELECT id, name FROM {project.test_schema}.basic_model ORDER BY id",
             fetch="all",
@@ -216,21 +239,21 @@ def test_single_quotes_are_doubled_and_survive(self, project, _run_all):
         """Quotes are doubled in the emitted SQL, and the remote literal
         round-trips to the value it's."""
         sql = _find_compiled_sql(project, "quotes_model.sql")
-        assert "OPENQUERY(\"LOCALLOOP\", 'SELECT ''it''''s'' AS msg')" in sql
+        assert _for_this_worker("OPENQUERY(\"LOCALLOOP\", 'SELECT ''it''''s'' AS msg')") in sql
         rows = project.run_sql(f"SELECT msg FROM {project.test_schema}.quotes_model", fetch="all")
         assert [row[0] for row in rows] == ["it's"]
 
     def test_carriage_returns_are_stripped_and_query_runs(self, project, _run_all):
         sql = _find_compiled_sql(project, "cr_model.sql")
         assert "\r" not in sql
-        assert 'OPENQUERY("LOCALLOOP", \'SELECT 1' in sql
+        assert _for_this_worker('OPENQUERY("LOCALLOOP", \'SELECT 1') in sql
         rows = project.run_sql(f"SELECT id FROM {project.test_schema}.cr_model", fetch="all")
         assert [row[0] for row in rows] == [1]
 
     def test_max_length_boundary_compiles(self, project, _run_all):
         """Exactly 8000 escaped characters is allowed."""
         sql = _find_compiled_sql(project, "max_length_model.sql")
-        assert 'OPENQUERY("LOCALLOOP", \'SELECT ' in sql
+        assert _for_this_worker('OPENQUERY("LOCALLOOP", \'SELECT ') in sql
 
     @pytest.mark.parametrize(
         "model_name,expected",

tests/functional/adapter/mssql/test_table_refresh_method.py

@@ -605,3 +605,164 @@ def test_column_order_mismatch_inserts_by_name(self, project):
 
         # Scratch cleaned up
         assert not table_exists(project, "dml_reorder_model__dbt_refresh")
+
+
+# -- Test: the columnstore survives the schema-change rename-swap fallback --
+
+dml_cci_schema_change_sql = """
+{{
+  config({
+    "materialized": "table",
+    "table_refresh_method": "dml"
+  })
+}}
+select 1 as id, 'hello' as val, 42 as new_col
+"""
+
+
+class TestDmlRefreshColumnstoreSurvivesSchemaChange:
+    """A schema change pushes the DML refresh onto its rename-swap fallback,
+    which renames the scratch table into position. The scratch came from
+    SELECT * INTO, which copies no index, and create_indexes only builds what
+    the `indexes` config names - never the as_columnstore CCI. So the model
+    came back a heap and stayed one, since every later run matched the new
+    schema and took the DELETE+INSERT path. Rebuilding the scratch through
+    create_table_as is what carries the columnstore across the swap.
+    """
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {"dml_cci_swap_model.sql": dml_with_columnstore_sql}
+
+    def test_columnstore_survives_schema_change(self, project):
+        results = run_dbt(["run"])
+        assert len(results) == 1
+        assert results[0].status == "success"
+        assert has_columnstore_index(project, "dml_cci_swap_model")
+
+        # Add a column: the refresh cannot swap by DELETE+INSERT any more and
+        # falls back to rename-swap.
+        write_model(project, "dml_cci_swap_model.sql", dml_cci_schema_change_sql)
+
+        results = run_dbt(["run"])
+        assert len(results) == 1
+        assert results[0].status == "success"
+
+        assert "new_col" in get_column_names(project, "dml_cci_swap_model")
+        assert has_columnstore_index(project, "dml_cci_swap_model")
+        assert not table_exists(project, "dml_cci_swap_model__dbt_refresh")
+
+
+# -- Test: NOT NULL survives the schema-change rename-swap fallback --
+
+dml_not_null_model_sql = """
+{{
+  config({
+    "materialized": "table",
+    "table_refresh_method": "dml",
+    "as_columnstore": False
+  })
+}}
+select try_cast('1' as int) as id, try_cast('hello' as varchar(5)) as val
+"""
+
+# Adds a column, which is what pushes the DML refresh onto its rename-swap
+# fallback.
+dml_not_null_model_wider_sql = """
+{{
+  config({
+    "materialized": "table",
+    "table_refresh_method": "dml",
+    "as_columnstore": False
+  })
+}}
+select try_cast('1' as int) as id,
+       try_cast('hello' as varchar(5)) as val,
+       try_cast('x' as varchar(5)) as extra
+"""
+
+dml_not_null_schema_yml = """
+version: 2
+models:
+  - name: dml_not_null_model
+    config:
+      contract:
+        enforced: true
+    columns:
+      - name: id
+        data_type: int
+        constraints:
+          - type: not_null
+      - name: val
+        data_type: varchar(5)
+        constraints:
+          - type: not_null
+"""
+
+dml_not_null_schema_wider_yml = (
+    dml_not_null_schema_yml
+    + """      - name: extra
+        data_type: varchar(5)
+"""
+)
+
+
+def get_not_null_columns(project, table_name):
+    """Get the names of a table's NOT NULL columns, in order."""
+    sql = (
+        f"SELECT COLUMN_NAME FROM INFORMATION_SCHEMA.COLUMNS "
+        f"WHERE TABLE_SCHEMA = '{project.test_schema}' "
+        f"AND TABLE_NAME = '{table_name}' "
+        f"AND IS_NULLABLE = 'NO' "
+        f"ORDER BY ORDINAL_POSITION"
+    )
+    with get_connection(project.adapter):
+        _, table = project.adapter.execute(sql, fetch=True)
+    return [row[0] for row in table.rows]
+
+
+class TestDmlRefreshNotNullSurvivesSchemaChange:
+    """The other half of the same bug as
+    TestDmlRefreshColumnstoreSurvivesSchemaChange: SELECT * INTO takes each
+    column's nullability from the query rather than from the contract, so a
+    schema change used to rename a table whose NOT NULLs had all become
+    nullable - and no later run put them back, every one of them matching the
+    new schema and taking the DELETE+INSERT path.
+
+    The model selects through try_cast so the query's own columns are nullable.
+    A literal would be inferred NOT NULL and carry across on its own, which
+    would pass whether or not the contract was honoured.
+    """
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "dml_not_null_model.sql": dml_not_null_model_sql,
+            "schema.yml": dml_not_null_schema_yml,
+        }
+
+    def test_not_null_survives_schema_change(self, project):
+        results = run_dbt(["run"])
+        assert len(results) == 1
+        assert results[0].status == "success"
+        assert get_not_null_columns(project, "dml_not_null_model") == ["id", "val"]
+
+        # A steady-state refresh keeps the table object, so it keeps them too.
+        results = run_dbt(["run"])
+        assert len(results) == 1
+        assert results[0].status == "success"
+        assert get_not_null_columns(project, "dml_not_null_model") == ["id", "val"]
+
+        # Add a column: the refresh cannot swap by DELETE+INSERT any more and
+        # falls back to rename-swap.
+        write_model(project, "dml_not_null_model.sql", dml_not_null_model_wider_sql)
+        write_model(project, "schema.yml", dml_not_null_schema_wider_yml)
+
+        results = run_dbt(["run"])
+        assert len(results) == 1
+        assert results[0].status == "success"
+
+        assert "extra" in get_column_names(project, "dml_not_null_model")
+        # extra is declared without a not_null constraint, so it must stay out.
+        assert get_not_null_columns(project, "dml_not_null_model") == ["id", "val"]
+        assert not table_exists(project, "dml_not_null_model__dbt_refresh")