diff --git a/CHANGELOG.md b/CHANGELOG.md index f7b18c7e..97f81685 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,7 +16,10 @@ #### Bugfixes +- Fix `dbt_sqlserver_use_dbt_transactions: True` (opt-in in 1.11, the default from 1.12) breaking two things that assumed the old autocommit-per-statement behavior: building an index with `build_options: {online: true}` or `{resumable: true}` on a fresh table or `table_refresh_method`'s rename-swap path (SQL Server rejects `RESUMABLE` inside a user transaction outright — `create_indexes` didn't split these out the way index reconciliation already did); and the `dbt_full_refresh_incomplete` crash marker on `full_refresh_build=prebuilt` and incremental full refreshes, which previously rode the same ambient transaction as the rebuild it's meant to survive, so a real failure mid-rebuild rolled the marker back along with everything else instead of leaving it in place to block the next normal run. +- Fix concurrent incremental models deadlocking (error 1205) at `threads` > 1 with `dbt_sqlserver_use_dbt_transactions: True` (opt-in in 1.11, the default from 1.12). The temp-relation build — `CREATE OR ALTER VIEW …__dbt_tmp_vw`, `SELECT * INTO …__dbt_tmp`, `DROP VIEW` — is all system-catalog DDL, and it was running inside the materialization's ambient transaction, so its exclusive `sys.sysschobjs` key locks were held until commit while a second worker's name resolution asked for shared locks on the same rows. Nothing in that build needs a transaction (the relation is throwaway); it was being opened accidentally, by a read-only catalog lookup that used `statement()`'s default `auto_begin=True`. The temp build and the fresh-create/full-refresh `create_table_as` batch now autocommit statement by statement and release their catalog locks as they go, while the incremental strategy DML stays transactional through to `adapter.commit()` as before. Pre-hook rollback semantics are unchanged: nothing is committed early, dbt only declines to open a transaction for a read. Because the temp build now commits standalone, a crashed run can leave a `__dbt_tmp` table behind, so the `SELECT * INTO` path drops adapter-generated throwaways first instead of failing the next run with `Msg 2714`; a fresh create of a real target still surfaces `Msg 2714` rather than silently destroying an object dbt does not know about. - Fix `drop_all_indexes_on_table()` and its component macros (`drop_pk_constraints()`, `drop_fk_constraints()`, `drop_xml_indexes()`, `drop_spatial_indexes()`) matching tables by name only. A post-hook on a model dropped the primary key, inbound foreign keys, and XML/spatial indexes of every same-named table in *other* schemas of the same database. All four now filter on the model's schema as well. Identifiers in the generated dynamic SQL are also wrapped in `QUOTENAME()`. +- Apply the existing `information_schema_hints()` (`with (nolock)`) to the read-only catalog lookups that were missing it: `get_relation_last_modified` (reached by `dbt source freshness` on a source with no `loaded_at_field`), the foreign-key and primary-key introspection in `drop_fk_constraints` / `drop_pk_constraints` / `drop_fk_references`, `list_nonclustered_rowstore_indexes`, and `find_references`. SQL Server holds transaction-scoped exclusive locks on the system catalog for the duration of any `CREATE`/`ALTER`/`DROP`, so an un-hinted catalog read from dbt waits on every other writer in the database — a second dbt project, an ETL tool, a DBA mid-deployment — and since the default `LOCK_TIMEOUT` is `-1` it waits indefinitely, with dbt appearing to hang and no error. Catalog reads that act as correctness guards rather than discovery (the schema-exists check before `CREATE SCHEMA`, the principal-exists check in `apply_grants`, and the `dbt_full_refresh_incomplete` extended-property marker) are deliberately left un-hinted, since a dirty read there would weaken the guard. - **Behavior change:** `SET XACT_ABORT ON` is now a session-level default on every connection the adapter opens, fixing silent, committed data loss on the DML table refresh path (and any other multi-statement batch): a mid-batch run-time error (e.g. a `NOT NULL`/constraint violation on the swap `INSERT`) previously aborted only the failing statement, not the batch, so a trailing `COMMIT` could still commit a partial result — most notably an emptied target after the `DELETE` succeeded and the `INSERT` failed. This is on by default starting in this release; opt out per profile with `xact_abort: false` if you rely on continue-on-error batch semantics in custom hooks. [#718](https://github.com/dbt-msft/dbt-sqlserver/issues/718) #### Under the hood diff --git a/dbt/adapters/sqlserver/sqlserver_adapter.py b/dbt/adapters/sqlserver/sqlserver_adapter.py index a782bddb..2d443992 100644 --- a/dbt/adapters/sqlserver/sqlserver_adapter.py +++ b/dbt/adapters/sqlserver/sqlserver_adapter.py @@ -419,6 +419,57 @@ def expand_target_column_types( def parse_index(self, raw_index: Any) -> Optional[SQLServerIndexConfig]: return SQLServerIndexConfig.parse(raw_index) + @available + def index_needs_own_batch(self, raw_index: Any) -> bool: + """True when raw_index's build_options (ONLINE / RESUMABLE) force its + CREATE INDEX to run outside any transaction: SQL Server rejects + RESUMABLE inside a user transaction (error 574), and an ONLINE build + wrapped in one holds its locks until commit, negating the point.""" + parsed = self.parse_index(raw_index) + if not parsed: + return False + return create_needs_own_batch(parsed.build_options) + + @available + def commit_if_open(self) -> None: + """Commit the current transaction if one is open - a no-op otherwise. + + dbt_sqlserver_use_dbt_transactions on (default) wraps a + materialization's whole build, from its first statement through its + own trailing ``adapter.commit()``, in one continuous ambient + transaction. Some statements must not share that transaction with + whatever runs after them: an ONLINE/RESUMABLE index build (SQL Server + rejects RESUMABLE inside a user transaction outright, and ONLINE + holds its locks until commit either way), or a full-refresh-in- + progress marker, which exists specifically to survive a later + failure and so must not roll back with it. This makes the prior + statement durable on its own; pair with begin_if_closed once the + statement(s) that must run outside a transaction are done, to leave + later code (more such statements, or the materialization's own + trailing ``adapter.commit()``, which raises if it finds nothing open) + working as if this call had never happened. + + A no-op when no transaction is open: the caller's statement already + ran autocommitted on its own (e.g. via ``run_query``'s + ``auto_begin=false``, before anything else began one), so there is + nothing to flush. Also a no-op, at the SQL level, whenever + dbt_sqlserver_use_dbt_transactions is off: begin/commit still flip + dbt-core's bookkeeping (see + SQLServerConnectionManager.add_begin_query/add_commit_query), but + emit no real T-SQL, matching the driver's own autocommit. + """ + connection = self.connections.get_thread_connection() + if connection is not None and connection.transaction_open: + self.connections.commit() + + @available + def begin_if_closed(self) -> None: + """Begin a transaction if none is open - a no-op otherwise. See + commit_if_open, which this pairs with.""" + connection = self.connections.get_thread_connection() + if connection is not None and not connection.transaction_open: + self.connections.begin() + @available def validate_indexes( self, raw_indexes: Any, as_columnstore: Any = False, drop_unmanaged: Any = False diff --git a/dbt/include/sqlserver/macros/adapters/indexes.sql b/dbt/include/sqlserver/macros/adapters/indexes.sql index 7c1b1c36..0eb3425e 100644 --- a/dbt/include/sqlserver/macros/adapters/indexes.sql +++ b/dbt/include/sqlserver/macros/adapters/indexes.sql @@ -65,8 +65,8 @@ declare @drop_fk_constraints nvarchar(max); select @drop_fk_constraints = ( select 'IF OBJECT_ID(''' + REPLACE(QUOTENAME(SCHEMA_NAME(sys.foreign_keys.[schema_id])) + '.' + QUOTENAME(sys.foreign_keys.[name]), '''', '''''') + ''', ''F'') IS NOT NULL ALTER TABLE ' + QUOTENAME(SCHEMA_NAME(sys.foreign_keys.[schema_id])) + '.' + QUOTENAME(OBJECT_NAME(sys.foreign_keys.[parent_object_id])) + ' DROP CONSTRAINT ' + QUOTENAME(sys.foreign_keys.[name]) + ';' - from sys.foreign_keys - inner join sys.tables on sys.foreign_keys.[referenced_object_id] = sys.tables.[object_id] + from sys.foreign_keys {{ information_schema_hints() }} + inner join sys.tables {{ information_schema_hints() }} on sys.foreign_keys.[referenced_object_id] = sys.tables.[object_id] where SCHEMA_NAME(sys.tables.[schema_id]) = '{{ this.schema }}' and sys.tables.[name] = '{{ this.table }}' for xml path('') @@ -90,8 +90,8 @@ declare @drop_pk_constraints nvarchar(max); select @drop_pk_constraints = ( select 'IF INDEXPROPERTY(' + CONVERT(VARCHAR(MAX), sys.tables.[object_id]) + ', ' + QUOTENAME(sys.indexes.[name], '''') + ', ''IndexId'') IS NOT NULL ALTER TABLE ' + QUOTENAME(SCHEMA_NAME(sys.tables.[schema_id])) + '.' + QUOTENAME(sys.tables.[name]) + ' DROP CONSTRAINT ' + QUOTENAME(sys.indexes.[name]) + ';' - from sys.indexes - inner join sys.tables on sys.indexes.[object_id] = sys.tables.[object_id] + from sys.indexes {{ information_schema_hints() }} + inner join sys.tables {{ information_schema_hints() }} on sys.indexes.[object_id] = sys.tables.[object_id] where sys.indexes.is_primary_key = 1 and SCHEMA_NAME(sys.tables.[schema_id]) = '{{ this.schema }}' and sys.tables.[name] = '{{ this.table }}' @@ -187,18 +187,18 @@ col1.name AS [column], tab2.name AS [referenced_table], col2.name AS [referenced_column] - FROM sys.foreign_key_columns fkc - INNER JOIN sys.objects obj + FROM sys.foreign_key_columns fkc {{ information_schema_hints() }} + INNER JOIN sys.objects obj {{ information_schema_hints() }} ON obj.object_id = fkc.constraint_object_id - INNER JOIN sys.tables tab1 + INNER JOIN sys.tables tab1 {{ information_schema_hints() }} ON tab1.object_id = fkc.parent_object_id - INNER JOIN sys.schemas sch + INNER JOIN sys.schemas sch {{ information_schema_hints() }} ON tab1.schema_id = sch.schema_id - INNER JOIN sys.columns col1 + INNER JOIN sys.columns col1 {{ information_schema_hints() }} ON col1.column_id = parent_column_id AND col1.object_id = tab1.object_id - INNER JOIN sys.tables tab2 + INNER JOIN sys.tables tab2 {{ information_schema_hints() }} ON tab2.object_id = fkc.referenced_object_id - INNER JOIN sys.columns col2 + INNER JOIN sys.columns col2 {{ information_schema_hints() }} ON col2.column_id = referenced_column_id AND col2.object_id = tab2.object_id WHERE sch.name = '{{ relation.schema }}' and tab2.name = '{{ relation.identifier }}' {% endcall %} @@ -216,8 +216,8 @@ SELECT i.name AS index_name , i.name + '__dbt_backup' as index_new_name , COL_NAME(ic.object_id,ic.column_id) AS column_name - FROM sys.indexes AS i - INNER JOIN sys.index_columns AS ic + FROM sys.indexes AS i {{ information_schema_hints() }} + INNER JOIN sys.index_columns AS ic {{ information_schema_hints() }} ON i.object_id = ic.object_id AND i.index_id = ic.index_id and i.type <> 5 WHERE i.object_id = OBJECT_ID('{{ relation.schema }}.{{ relation.identifier }}') @@ -226,18 +226,18 @@ SELECT obj.name AS index_name , obj.name + '__dbt_backup' as index_new_name , col1.name AS column_name - FROM sys.foreign_key_columns fkc - INNER JOIN sys.objects obj + FROM sys.foreign_key_columns fkc {{ information_schema_hints() }} + INNER JOIN sys.objects obj {{ information_schema_hints() }} ON obj.object_id = fkc.constraint_object_id - INNER JOIN sys.tables tab1 + INNER JOIN sys.tables tab1 {{ information_schema_hints() }} ON tab1.object_id = fkc.parent_object_id - INNER JOIN sys.schemas sch + INNER JOIN sys.schemas sch {{ information_schema_hints() }} ON tab1.schema_id = sch.schema_id - INNER JOIN sys.columns col1 + INNER JOIN sys.columns col1 {{ information_schema_hints() }} ON col1.column_id = parent_column_id AND col1.object_id = tab1.object_id - INNER JOIN sys.tables tab2 + INNER JOIN sys.tables tab2 {{ information_schema_hints() }} ON tab2.object_id = fkc.referenced_object_id - INNER JOIN sys.columns col2 + INNER JOIN sys.columns col2 {{ information_schema_hints() }} ON col2.column_id = referenced_column_id AND col2.object_id = tab2.object_id WHERE sch.name = '{{ relation.schema }}' and tab1.name = '{{ relation.identifier }}' @@ -453,12 +453,51 @@ {%- do adapter.validate_indexes( raw_indexes, as_columnstore, config.get('drop_unmanaged_indexes', default=false) ) -%} + {%- set creates_no_txn = [] -%} {%- for _index_dict in raw_indexes %} {%- set create_index_sql = get_create_index_sql(relation, _index_dict) -%} {% if create_index_sql %} - {% do run_query(create_index_sql) %} + {% if adapter.index_needs_own_batch(_index_dict) %} + {%- do creates_no_txn.append(_index_dict) -%} + {% else %} + {% do run_query(create_index_sql) %} + {% endif %} {% endif %} {%- endfor %} + {% do sqlserver__create_indexes_no_txn(relation, creates_no_txn) %} +{% endmacro %} + + +{% macro sqlserver__create_indexes_no_txn(relation, index_dicts) %} + {#- + Build ONLINE / RESUMABLE indexes as standalone (non-transactional) + statements. SQL Server forbids RESUMABLE in a user transaction, and an + ONLINE build wrapped in one holds its locks until commit, negating the + non-blocking intent. + + With dbt_sqlserver_use_dbt_transactions on (default), the + materialization's whole build - from the first statement through its own + trailing adapter.commit() - runs inside one continuous ambient + transaction, so run_query's auto_begin=false is not enough on its own to + escape it: commit it first (see adapter.commit_if_open), before ANY of + these run - committing between them would just reopen one and land the + next create right back inside it - then reopen once after the whole + batch (adapter.begin_if_closed), which always leaves one open afterward + even if none was open before, so later code (grants, persist_docs, the + materialization's own trailing adapter.commit(), which raises if it + finds nothing open) sees one exactly as it would without this macro. + When the flag is off, or no earlier statement this run has opened one, + commit_if_open is a no-op and these still run standalone via run_query's + auto_begin=false. + -#} + {% if index_dicts %} + {{ adapter.commit_if_open() }} + {% for index_dict in index_dicts %} + {% do log("Creating ONLINE/RESUMABLE index outside the transaction on " ~ relation, info=true) %} + {% do run_query(get_create_index_sql(relation, index_dict)) %} + {% endfor %} + {{ adapter.begin_if_closed() }} + {% endif %} {% endmacro %} @@ -500,15 +539,13 @@ ~ ";\ncommit transaction;" ) %} {% endif %} - {#- ONLINE / RESUMABLE creates cannot run inside the transaction above - (SQL Server forbids RESUMABLE in a user transaction, and an ONLINE build - wrapped in one holds its locks until commit, negating the non-blocking - intent). Apply them individually in autocommit. The drops above have - already committed, so a replacement index still builds after its - predecessor is gone; these are not part of the atomic batch, so there is - a brief window where the new index is absent for readers. -#} - {%- for index_dict in result['creates_no_txn'] %} - {% do log("Creating ONLINE/RESUMABLE index outside the reconcile transaction on " ~ relation, info=true) %} - {% do run_query(sqlserver__get_create_index_sql(relation, index_dict)) %} - {%- endfor %} + {#- ONLINE / RESUMABLE creates cannot run inside the transaction above, nor + inside the ambient dbt-managed transaction that wraps the whole + materialization when dbt_sqlserver_use_dbt_transactions is on - see + sqlserver__create_indexes_no_txn. The drops (and any transactional + creates) above have already committed by the time these run, so a + replacement index still builds after its predecessor is gone; these are + not part of the atomic batch, so there is a brief window where the new + index is absent for readers. -#} + {% do sqlserver__create_indexes_no_txn(relation, result['creates_no_txn']) %} {% endmacro %} diff --git a/dbt/include/sqlserver/macros/adapters/metadata.sql b/dbt/include/sqlserver/macros/adapters/metadata.sql index a7237708..b1e2258d 100644 --- a/dbt/include/sqlserver/macros/adapters/metadata.sql +++ b/dbt/include/sqlserver/macros/adapters/metadata.sql @@ -194,8 +194,8 @@ , s.name as [schema] , o.modify_date as last_modified , current_timestamp as snapshotted_at - from sys.objects o - inner join sys.schemas s on o.schema_id = s.schema_id and [type] = 'U' + from sys.objects o {{ information_schema_hints() }} + inner join sys.schemas s {{ information_schema_hints() }} on o.schema_id = s.schema_id and [type] = 'U' where ( {%- for relation in relations -%} (upper(s.name) = upper('{{ relation.schema }}') and diff --git a/dbt/include/sqlserver/macros/adapters/relation.sql b/dbt/include/sqlserver/macros/adapters/relation.sql index 6d7a7079..a248b611 100644 --- a/dbt/include/sqlserver/macros/adapters/relation.sql +++ b/dbt/include/sqlserver/macros/adapters/relation.sql @@ -8,15 +8,21 @@ {% macro sqlserver__get_drop_sql(relation) -%} {% if relation.type == 'view' -%} - {% call statement('find_references', fetch_result=true) %} + {#- auto_begin=false, matching dbt-adapters' own relations/drop.sql: a + read-only lookup must not open the ambient transaction. This is the + first statement of the temp-relation build (create.sql -> + adapter.drop_relation), so with the default auto_begin=True it held + that build's sys.sysschobjs X locks until the materialization + committed, deadlocking a concurrent worker (error 1205). -#} + {% call statement('find_references', fetch_result=true, auto_begin=false) %} {{ get_use_database_sql(relation.database) }} select sch.name as schema_name, obj.name as view_name - from sys.sql_expression_dependencies refs - inner join sys.objects obj + from sys.sql_expression_dependencies refs {{ information_schema_hints() }} + inner join sys.objects obj {{ information_schema_hints() }} on refs.referencing_id = obj.object_id - inner join sys.schemas sch + inner join sys.schemas sch {{ information_schema_hints() }} on obj.schema_id = sch.schema_id where refs.referenced_database_name = '{{ relation.database }}' and refs.referenced_schema_name = '{{ relation.schema }}' diff --git a/dbt/include/sqlserver/macros/materializations/hooks.sql b/dbt/include/sqlserver/macros/materializations/hooks.sql index 0f50ed0b..8da27177 100644 --- a/dbt/include/sqlserver/macros/materializations/hooks.sql +++ b/dbt/include/sqlserver/macros/materializations/hooks.sql @@ -5,7 +5,11 @@ {% if not adapter.behavior.dbt_sqlserver_use_dbt_transactions %} if @@trancount > 0 commit; -- post hooks after fictitious transaction work as expected {% else %} - commit; -- align transaction=False hook behavior with dbt-core transaction semantics. + {#- guarded, not a bare COMMIT: nothing guarantees a transaction is + open here. A bare one appeared safe only because some earlier + statement had auto-begun one (find_references, until relation.sql + stopped doing that); with @@TRANCOUNT = 0 it raises Msg 3902. -#} + if @@trancount > 0 commit; -- align transaction=False hook behavior with dbt-core transaction semantics. {% endif %} {% endcall %} {% endif %} diff --git a/dbt/include/sqlserver/macros/materializations/models/incremental/incremental.sql b/dbt/include/sqlserver/macros/materializations/models/incremental/incremental.sql index 82fcadd6..d3c10736 100644 --- a/dbt/include/sqlserver/macros/materializations/models/incremental/incremental.sql +++ b/dbt/include/sqlserver/macros/materializations/models/incremental/incremental.sql @@ -30,14 +30,24 @@ {{ run_hooks(pre_hooks, inside_transaction=True) }} {% set to_drop = [] %} + {% set prebuilt_handled = false %} + {#- true only where the statement('main') batch below carries create_table_as + DDL, i.e. the fresh-create / full-refresh branches. The incremental + branch's strategy DML stays transactional, so it leaves this false. -#} + {% set build_sql_is_create_table_as = false %} {% if existing_relation is none %} {% if config.get('full_refresh_build', 'heap_then_index') == 'prebuilt' %} - {#- first build: load straight into the clustered design -#} - {% set build_sql = sqlserver__create_table_as_prebuilt(target_relation, sql) %} + {#- first build: load straight into the clustered design. Calls its own + statement() blocks (including 'main') rather than returning SQL to + run below, so it can commit its in-progress marker independently of + the load that follows - see the macro for why. -#} + {% do sqlserver__create_table_as_prebuilt(target_relation, sql) %} {% set prebuilt_cache_add = true %} + {% set prebuilt_handled = true %} {% else %} {% set build_sql = get_create_table_as_sql(False, target_relation, sql) %} + {% set build_sql_is_create_table_as = true %} {% endif %} {% elif full_refresh_mode %} {#- the target is marked as having a full refresh in flight (blocking @@ -59,10 +69,12 @@ {% do sqlserver__mark_full_refresh_incomplete(existing_relation) %} {% endif %} {% do adapter.drop_relation(existing_relation) %} - {% set build_sql = sqlserver__create_table_as_prebuilt(target_relation, sql) %} + {% do sqlserver__create_table_as_prebuilt(target_relation, sql) %} {% set prebuilt_cache_add = true %} + {% set prebuilt_handled = true %} {% else %} {% set build_sql = get_create_table_as_sql(False, intermediate_relation, sql) %} + {% set build_sql_is_create_table_as = true %} {% if existing_relation.type == 'table' %} {% do sqlserver__mark_full_refresh_incomplete(existing_relation) %} {% endif %} @@ -76,6 +88,14 @@ {% do sqlserver__assert_no_incomplete_full_refresh(existing_relation) %} {% endif %} + {#- The temp build is all catalog DDL (CREATE OR ALTER VIEW / SELECT * + INTO / DROP VIEW) and must not share the ambient transaction with the + strategy DML: held to commit, its sysschobjs X keylocks deadlock a + second worker. Nothing opens a transaction here - run_query never + auto-begins, and find_references (relation.sql) no longer does either - + so each statement autocommits and drops its catalog locks as it + finishes. The strategy DML below still runs transactionally, via + statement('main')'s default auto_begin through to adapter.commit(). -#} {% do run_query(get_create_table_as_sql(True, temp_relation, sql)) %} {% set contract_config = config.get('contract') %} @@ -102,9 +122,30 @@ {% do to_drop.append(temp_relation) %} {% endif %} - {% call statement("main") %} - {{ build_sql }} - {% endcall %} + {% if not prebuilt_handled %} + {% if build_sql_is_create_table_as %} + {#- Same reason as the temp build above: this batch is create_table_as + catalog DDL, so letting statement() open the ambient transaction + would hold its sysschobjs X keylocks until adapter.commit() and + deadlock a second worker. -#} + {% call statement("main", auto_begin=False) %} + {{ build_sql }} + {% endcall %} + {% else %} + {% call statement("main") %} + {{ build_sql }} + {% endcall %} + {% endif %} + {% if build_sql_is_create_table_as %} + {#- Reopen the ambient transaction the batch above declined to start, + so the swap and the tail (grants/persist_docs/masks/indexes/ + post-hooks) keep their semantics and adapter.commit() below has a + matching BEGIN rather than raising Msg 3902. No-op when the flag is + off. -#} + {% do adapter.commit_if_open() %} + {% do adapter.begin_if_closed() %} + {% endif %} + {% endif %} {% if need_swap %} {% do adapter.rename_relation(target_relation, backup_relation) %} diff --git a/dbt/include/sqlserver/macros/materializations/models/table/table.sql b/dbt/include/sqlserver/macros/materializations/models/table/table.sql index b1cea241..5a58bba7 100644 --- a/dbt/include/sqlserver/macros/materializations/models/table/table.sql +++ b/dbt/include/sqlserver/macros/materializations/models/table/table.sql @@ -68,9 +68,11 @@ {% endif %} {% endif %} - {% call statement('main') -%} - {{ sqlserver__create_table_as_prebuilt(target_relation, sql) }} - {%- endcall %} + {#- create_table_as_prebuilt issues its own statement() calls (including + 'main') rather than being wrapped in one here, so it can commit its + in-progress marker independently of the load that follows - see the + macro for why. -#} + {% do sqlserver__create_table_as_prebuilt(target_relation, sql) %} {#- the prebuilt path lands the table via raw SQL, not a cache-maintaining adapter method (rename_relation), and above it may have dropped the diff --git a/dbt/include/sqlserver/macros/relations/table/create.sql b/dbt/include/sqlserver/macros/relations/table/create.sql index ec5ac329..be74c2a7 100644 --- a/dbt/include/sqlserver/macros/relations/table/create.sql +++ b/dbt/include/sqlserver/macros/relations/table/create.sql @@ -9,6 +9,19 @@ {%- endif -%} {%- set tmp_relation = relation.incorporate(path={"identifier": relation.identifier ~ '__dbt_tmp_vw'}, type='view') -%} + {#- Now that the incremental temp build commits standalone (see + incremental.sql), a crash can leave a throwaway table behind and the + SELECT * INTO below would hit Msg 2714. Drop it first, but only for + adapter-generated throwaways: `temporary` covers the incremental + __dbt_temp build, the suffix covers the full-refresh / table-refresh + __dbt_tmp intermediate. Suffix match is exact, never substring, so a + user model named stg__dbt_tmp_x is untouched. + Never guard a fresh-create of the real target: dbt has decided that + table does not exist, so 2714 must still surface rather than silently + destroying an object dbt does not know about. -#} + {%- set _ident = relation.identifier -%} + {%- set build_into_temp = temporary or _ident.endswith('__dbt_tmp') or _ident.endswith('__dbt_tmp_vw') -%} + {%- do adapter.drop_relation(tmp_relation) -%} USE [{{ relation.database }}]; {{ get_create_view_as_sql(tmp_relation, sql) }} @@ -33,6 +46,10 @@ SELECT {{listColumns}} FROM {{tmp_relation}} {{ query_label }} {% else %} + {%- if build_into_temp -%} + IF OBJECT_ID('{{ escape_single_quotes(relation.schema) }}.{{ escape_single_quotes(relation.identifier) }}', 'U') IS NOT NULL + EXEC('DROP TABLE {{ relation }}'); + {%- endif -%} SELECT * INTO {{ table_name }} FROM {{ tmp_relation }} {{ query_label }} {% endif %} {%- endset -%} @@ -92,32 +109,57 @@ {%- endif -%} {%- do adapter.drop_relation(tmp_relation) -%} - USE [{{ relation.database }}]; - {{ get_create_view_as_sql(tmp_relation, sql) }} - {#- drop any physical target before the bare CREATE / SELECT ... INTO below. - OBJECT_ID reads the database directly, so this is robust to dbt's - relation cache being stale (reporting none while the table exists) - - which would otherwise collide with Msg 2714 (object already exists) -#} - IF OBJECT_ID('{{ escape_single_quotes(relation.schema) }}.{{ escape_single_quotes(relation.identifier) }}', 'U') IS NOT NULL - EXEC('DROP TABLE {{ relation }}'); + {#- Setup runs as its own statement (named 'main', so dbt's compiled-SQL + artifacts and adapter_response still resolve normally), separate from + the load below: the extended-property marker set here exists to + survive a failed load, so it must not share a transaction with it. + With dbt_sqlserver_use_dbt_transactions on (default), the + materialization's whole build - including this statement - runs + inside one continuous ambient transaction; without the split below, a + later failure during the load would roll the marker back right along + with it, defeating the point. -#} + {%- set setup_sql -%} + USE [{{ relation.database }}]; + {{ get_create_view_as_sql(tmp_relation, sql) }} - {% if contract_enforced %} - {%- set ddl_query -%} - CREATE TABLE {{ relation }} - {{ get_assert_columns_equivalent(sql) }} - {{ build_columns_constraints(relation) }} - {%- endset -%} - EXEC('{{- escape_single_quotes(ddl_query) -}}') - {% else %} - EXEC('SELECT TOP 0 * INTO {{ relation }} FROM {{ tmp_relation }}') - {% endif %} + {#- drop any physical target before the bare CREATE / SELECT ... INTO + below. OBJECT_ID reads the database directly, so this is robust to + dbt's relation cache being stale (reporting none while the table + exists) - which would otherwise collide with Msg 2714 (object + already exists) -#} + IF OBJECT_ID('{{ escape_single_quotes(relation.schema) }}.{{ escape_single_quotes(relation.identifier) }}', 'U') IS NOT NULL + EXEC('DROP TABLE {{ relation }}'); - {# mark the rebuild in progress; removed atomically with the load below #} - EXEC sp_addextendedproperty @name = N'dbt_full_refresh_incomplete', @value = '1', - @level0type = N'SCHEMA', @level0name = N'{{ escape_single_quotes(relation.schema) }}', - @level1type = N'TABLE', @level1name = N'{{ escape_single_quotes(relation.identifier) }}' + {% if contract_enforced %} + {%- set ddl_query -%} + CREATE TABLE {{ relation }} + {{ get_assert_columns_equivalent(sql) }} + {{ build_columns_constraints(relation) }} + {%- endset -%} + EXEC('{{- escape_single_quotes(ddl_query) -}}') + {% else %} + EXEC('SELECT TOP 0 * INTO {{ relation }} FROM {{ tmp_relation }}') + {% endif %} + + {# mark the rebuild in progress; removed atomically with the load below #} + EXEC sp_addextendedproperty @name = N'dbt_full_refresh_incomplete', @value = '1', + @level0type = N'SCHEMA', @level0name = N'{{ escape_single_quotes(relation.schema) }}', + @level1type = N'TABLE', @level1name = N'{{ escape_single_quotes(relation.identifier) }}' + {%- endset %} + {% call statement('main') -%} + {{ setup_sql }} + {%- endcall %} + + {#- Commit the marker onto its own, then reopen so the load below (and + the rest of the materialization: grants, persist_docs, its own + trailing adapter.commit()) is unaffected. See adapter.commit_if_open / + begin_if_closed; 'main' above always opens a transaction (default + auto_begin), so this always actually commits here. -#} + {{ adapter.commit_if_open() }} + {{ adapter.begin_if_closed() }} + {%- set load_sql -%} {% if as_columnstore %} {{ sqlserver__create_clustered_columnstore_index(relation) }} {% elif prebuilt_ns.clustered_dict is not none %} @@ -156,7 +198,10 @@ EXEC('{{- escape_single_quotes(insert_query) -}}') EXEC('DROP VIEW IF EXISTS {{ tmp_relation.include(database=False) }}') - + {%- endset %} + {% call statement('create_table_as_prebuilt_load') -%} + {{ load_sql }} + {%- endcall %} {% endmacro %} @@ -172,6 +217,19 @@ ~ " @level0type = N'SCHEMA', @level0name = N'" ~ relation.schema ~ "'," ~ " @level1type = N'TABLE', @level1name = N'" ~ relation.identifier ~ "'" ) %} + {#- This marker exists to survive a failed rebuild, so it must not ride on + the same transaction as the rebuild it's guarding: if a prior + statement this run (e.g. a pre-hook) already opened the ambient + dbt-managed transaction, a later failure would roll the marker back + right along with it, defeating the point. Commit it, then reopen so + later code sees a transaction open exactly as it would without this + pair (begin_if_closed always leaves one open, whether or not + commit_if_open just found one to close) - commit_if_open alone is a + no-op when run_query above ran standalone, i.e. the common case of + this being the first statement of the run. See adapter.commit_if_open + / begin_if_closed. -#} + {{ adapter.commit_if_open() }} + {{ adapter.begin_if_closed() }} {%- endmacro %} diff --git a/tests/functional/adapter/dbt/test_concurrent_incremental.py b/tests/functional/adapter/dbt/test_concurrent_incremental.py new file mode 100644 index 00000000..20e52f85 --- /dev/null +++ b/tests/functional/adapter/dbt/test_concurrent_incremental.py @@ -0,0 +1,78 @@ +import pytest + +from dbt.cli.main import dbtRunner + +# Concurrency regression for the incremental materialization (plan +# 20260801-dbt-sqlserver-concurrent-incremental-deadlock). +# +# Root cause: with dbt_sqlserver_use_dbt_transactions on (the rc1 default), +# a read-only catalog lookup (find_references in sqlserver__get_drop_sql, +# the first statement of the temp build) auto-began the ambient transaction. +# The incremental temp build (create_table_as: CREATE OR ALTER VIEW + +# SELECT * INTO + DROP VIEW, all user-DB catalog DDL) then ran inside that +# transaction, holding transaction-scoped sysschobjs X keylocks until the +# final adapter.commit(). Two independent incremental models run in parallel +# (threads=2) then inverted X/S and deadlocked (Msg 1205). +# +# The fix stops the ambient transaction from being opened accidentally for +# catalog reads: find_references runs with auto_begin=false (relation.sql), +# and the fresh-create / full-refresh create_table_as batch runs with +# statement("main", auto_begin=False), so each statement autocommits and +# releases its catalog locks as it completes. The strategy DML keeps the +# default auto_begin and stays transactional until the final commit. +# +# Follows the BaseTransactionsEnabled pattern (test_transactions.py): the +# deadlock only forms when the flag wraps the build in the ambient +# transaction. Model SQL is values-derived (no sys.all_objects) to remove +# the catalog-scan confound flagged by the independent evaluation (A.1). +_concurrent_incremental_model_sql = """ +{{ config(materialized='incremental', unique_key='id') }} +select + row_number() over (order by (select null)) as id, + current_timestamp as loaded_at +from (values (1),(2),(3),(4),(5),(6),(7),(8),(9),(10)) as a(n) +cross join (values (1),(2),(3),(4),(5),(6),(7),(8),(9),(10)) as b(n) +cross join (values (1),(2),(3),(4),(5),(6),(7),(8),(9),(10)) as c(n) +cross join (values (1),(2),(3),(4),(5)) as d(n) +""" + + +class BaseConcurrentIncremental: + @pytest.fixture(scope="class") + def project_config_update(self): + return {"flags": {"dbt_sqlserver_use_dbt_transactions": True}} + + @pytest.fixture(scope="class") + def models(self): + return { + "model_a.sql": _concurrent_incremental_model_sql, + "model_b.sql": _concurrent_incremental_model_sql, + } + + def _build_concurrently(self, extra_args=None): + runner = dbtRunner() + result = runner.invoke(["build", "--threads", "2"] + (extra_args or [])) + assert result.success, ( + f"concurrent incremental build failed: {result.exception or result.result}" + ) + + def test_two_incremental_models_build_concurrently(self, project): + # Run 1: fresh-create path (create_table_as into the final target) for + # both models in parallel - covers the fresh-create deadlock class. + self._build_concurrently() + # Run 2: incremental temp-build + strategy-DML path - the boundary the + # fix releases early. + self._build_concurrently() + # Run 3: full-refresh path (create into __dbt_tmp intermediate + swap) + # for both models in parallel. + self._build_concurrently(["--full-refresh"]) + + for model in ("model_a", "model_b"): + rows = project.run_sql( + f"select count(*) as cnt from {project.test_schema}.{model}", fetch="one" + ) + assert rows[0] == 5000 + + +class TestConcurrentIncremental(BaseConcurrentIncremental): + pass diff --git a/tests/functional/adapter/mssql/test_catalog_nolock.py b/tests/functional/adapter/mssql/test_catalog_nolock.py new file mode 100644 index 00000000..664811ec --- /dev/null +++ b/tests/functional/adapter/mssql/test_catalog_nolock.py @@ -0,0 +1,129 @@ +"""The adapter's own catalog lookups must not queue behind unrelated DDL. + +SQL Server takes transaction-scoped X locks on the system catalog for every +CREATE/ALTER/DROP. Any *other* session that reads the catalog under the default +READ COMMITTED isolation has to wait for those X locks to be released - and the +default LOCK_TIMEOUT is -1, so it waits forever. + +That makes an un-hinted catalog query in the adapter a hard dependency on every +other writer in the database: a second dbt project, an ETL tool, a DBA running a +long deployment script. dbt appears to hang with no error. + +``information_schema_hints()`` (``with (nolock)``) is the existing fix for this; +it is already applied to ``list_relations_without_caching`` and friends. This +test covers ``get_relation_last_modified`` - reached by ``dbt source freshness`` +on a source with no ``loaded_at_field`` - which was missing it. + +The hint only ever affects the adapter's own read-only metadata discovery. It is +deliberately NOT applied to catalog reads that act as correctness guards +(schema-exists before CREATE SCHEMA, the extended-properties full-refresh +marker), where a dirty read would weaken the guard rather than just widen +discovery. +""" + +import threading +import time + +import pytest + +from dbt.tests.util import relation_from_name, run_dbt + +# Enough uncommitted objects that the blocker holds locks across the catalog +# pages the reader has to scan. A handful is not reliably enough. +BLOCKER_OBJECT_COUNT = 40 + +# Upper bound on how long the blocker keeps its transaction open. The reader is +# expected to finish in milliseconds; this exists only so a regression fails the +# test instead of hanging the suite forever. +BLOCKER_HOLD_SECONDS = 60 + +# Without this the reader would wait indefinitely (SQL Server's default +# LOCK_TIMEOUT is -1) and the failure would look like a hang rather than a +# regression. With it, a regression surfaces as Msg 1222. +READER_LOCK_TIMEOUT_MS = 5000 + +probe_model_sql = """ +{{ config(materialized='table') }} +select 1 as id +""" + + +class BlockingWriter: + """A second session holding uncommitted DDL in the test schema. + + Runs on its own thread so it gets its own dbt connection - the connection + manager keys connections by thread - and rolls back on exit, so the objects + it creates never actually exist. + """ + + def __init__(self, project): + self.project = project + self.ready = threading.Event() + self.release = threading.Event() + self.error = None + self._thread = threading.Thread(target=self._run, daemon=True) + + def _run(self): + try: + with self.project.adapter.connection_named("nolock_blocker"): + self.project.adapter.execute("begin transaction") + try: + for i in range(BLOCKER_OBJECT_COUNT): + self.project.adapter.execute( + f"create table {self.project.test_schema}.nolock_blocker_{i} " + f"(id int not null)" + ) + self.ready.set() + self.release.wait(timeout=BLOCKER_HOLD_SECONDS) + finally: + self.project.adapter.execute("rollback transaction") + except Exception as exc: # surfaced by the test, not swallowed + self.error = exc + finally: + self.ready.set() + + def __enter__(self): + self._thread.start() + assert self.ready.wait(timeout=BLOCKER_HOLD_SECONDS), ( + "blocker never opened its transaction" + ) + if self.error is not None: + raise AssertionError(f"blocker session failed: {self.error}") + return self + + def __exit__(self, *exc_info): + self.release.set() + self._thread.join(timeout=BLOCKER_HOLD_SECONDS) + return False + + +class TestCatalogLookupNotBlockedByExternalDdl: + @pytest.fixture(scope="class") + def models(self): + return {"probe_model.sql": probe_model_sql} + + def test_get_relation_last_modified_ignores_uncommitted_ddl(self, project): + run_dbt(["run"]) + relation = relation_from_name(project.adapter, "probe_model") + + with BlockingWriter(project): + with project.adapter.connection_named("nolock_reader"): + project.adapter.execute(f"set lock_timeout {READER_LOCK_TIMEOUT_MS}") + + started = time.time() + try: + project.adapter.execute_macro( + "get_relation_last_modified", + kwargs={"information_schema": None, "relations": [relation]}, + ) + except Exception as exc: + raise AssertionError( + "get_relation_last_modified blocked behind another session's " + f"uncommitted DDL after {time.time() - started:.2f}s: {exc}. " + "The sys.objects/sys.schemas reads need " + "{{ information_schema_hints() }}." + ) from exc + finally: + # dbt keeps named connections open for reuse, so restore the + # server default rather than leaking a timeout into later tests. + project.adapter.execute("set lock_timeout -1")