review: conditionally skip nested transaction in DML refresh when flag is on; add snapshot functional test

Author: axellpadilla

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

@@ -55,17 +55,18 @@
 
     {# Atomic DML swap — RCSI protects concurrent readers #}
     {# When dbt_sqlserver_use_dbt_transactions is off (default), autocommit #}
-    {# ensures only this explicit transaction exists. When the flag is on, #}
-    {# the statement call auto-begins an outer transaction first. SQL Server #}
-    {# increments @@TRANCOUNT for nested BEGIN TRANSACTION statements. #}
-    {# Inner COMMIT only decrements @@TRANCOUNT; durability occurs when the outermost #}
-    {# transaction commits. A ROLLBACK rolls back the full transaction. #}
+    {# means we need the explicit BEGIN/COMMIT. When the flag is on, dbt #}
+    {# already wraps the statement call in a transaction, so skip it. #}
     {% call statement('dml_refresh_swap') -%}
+      {% if not adapter.behavior.dbt_sqlserver_use_dbt_transactions %}
       BEGIN TRANSACTION;
+      {% endif %}
       DELETE FROM {{ target_relation }};
       INSERT INTO {{ target_relation }} ({{ column_list }})
         SELECT {{ column_list }} FROM {{ refresh_relation }};
+      {% if not adapter.behavior.dbt_sqlserver_use_dbt_transactions %}
       COMMIT TRANSACTION;
+      {% endif %}
     {%- endcall %}
 
     {# Cleanup scratch table #}

tests/functional/adapter/dbt/test_transactions.py

@@ -174,3 +174,48 @@ def test_failed_then_successful_run(self, project):
         rows = project.run_sql("select id from {schema}.good_model", fetch="all")
         assert len(rows) == 1
         assert rows[0][0] == 1
+
+
+_snapshot_seed_csv = """id,name,updated_at
+1,alice,2024-01-01 00:00:00
+2,bob,2024-01-01 00:00:00
+"""
+
+_snapshot_sql = """
+{% snapshot snap %}
+{{ config(
+    target_schema=schema,
+    unique_key='id',
+    strategy='timestamp',
+    updated_at='updated_at',
+) }}
+select * from {{ ref('snap_seed') }}
+{% endsnapshot %}
+"""
+
+
+class TestSnapshotTransactionsOn(BaseTransactionsEnabled):
+    @pytest.fixture(scope="class")
+    def seeds(self):
+        return {"snap_seed.csv": _snapshot_seed_csv}
+
+    @pytest.fixture(scope="class")
+    def snapshots(self):
+        return {"snap.sql": _snapshot_sql}
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {}
+
+    def test_snapshot_create_and_merge(self, project):
+        run_dbt(["seed"])
+        results = run_dbt(["snapshot"])
+        assert len(results) == 1
+        assert results[0].status == "success"
+
+        rows = project.run_sql("select count(*) from {schema}.snap", fetch="one")
+        assert rows[0] == 2
+
+        results = run_dbt(["snapshot"])
+        assert len(results) == 1
+        assert results[0].status == "success"