diff --git a/CHANGELOG.md b/CHANGELOG.md index 1ed19b52..45396730 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,12 @@ # Changelog +### v1.11.1 + +#### Bugfixes + +- Fix a `view` model silently skipping a rebuild when text was removed from the *start* of its body (e.g. deleting a leading comment or CTE). The skip test compared the stored definition against the model with `endswith()`, so any edit whose new body was a tail of the old one looked unchanged: `dbt run` reported `PASS` but the change never reached the database, and `--full-refresh` did not fix it. The header (`CREATE [OR ALTER] VIEW AS`) is now split off at its separating ` AS ` and the body compared exactly. The comparison also no longer lowercases or strips whitespace, both of which made genuinely different bodies (a string literal differing only in case, or any literal containing spaces) compare equal; where the definition cannot be parsed with certainty the view is rebuilt rather than skipped. [#807](https://github.com/dbt-msft/dbt-sqlserver/issues/807) +- Fix snapshots failing on their second and later runs with `Invalid object name '..._dbt_tmp'`, and contract-enforced models silently losing their in-transaction `pre_hook` writes. `get_column_schema_from_query` reads a query's column shape by executing it, then returned without fetching the rows or closing the cursor. Closing a cursor whose result set the server is still producing makes the driver cancel the request, and SQL Server answers that cancel by rolling back the open transaction, since every connection runs `SET XACT_ABORT ON` (#718). Nothing is raised for any of it, so the snapshot lost the staging table it had just built and failed against it a statement later. The probe now drains and closes its cursor, as does the row-count probe in `expand_column_types`. Only queries opening with a CTE were affected - anything else is wrapped as `select * from (...) where 1 = 0` by `sqlserver__get_empty_subquery_sql` and returns no rows - which is why snapshot staging queries (`with snapshot_query as ...`, both `check` and `timestamp` strategies) and CTE-headed contract models were the ones that broke. A CTE-headed probe is now also described with `sp_describe_first_result_set` rather than executed, so it no longer runs its query twice per build. [#809](https://github.com/dbt-msft/dbt-sqlserver/issues/809) + ### v1.11.0 #### Features diff --git a/dbt/adapters/sqlserver/__version__.py b/dbt/adapters/sqlserver/__version__.py index b6c30336..a7edaff6 100644 --- a/dbt/adapters/sqlserver/__version__.py +++ b/dbt/adapters/sqlserver/__version__.py @@ -1 +1 @@ -version = "1.11.0" +version = "1.11.1" diff --git a/dbt/adapters/sqlserver/sqlserver_adapter.py b/dbt/adapters/sqlserver/sqlserver_adapter.py index 2d443992..394fb909 100644 --- a/dbt/adapters/sqlserver/sqlserver_adapter.py +++ b/dbt/adapters/sqlserver/sqlserver_adapter.py @@ -1,3 +1,4 @@ +import re from typing import Any, Dict, List, Optional import agate @@ -26,16 +27,118 @@ index_config_changes, normalize_drop_unmanaged, ) +from dbt.adapters.sqlserver.sqlserver_auth import is_mssql_python_backend from dbt.adapters.sqlserver.sqlserver_column import SQLServerColumn, SQLServerColumnNative from dbt.adapters.sqlserver.sqlserver_configs import SQLServerConfigs -from dbt.adapters.sqlserver.sqlserver_connections import SQLServerConnectionManager +from dbt.adapters.sqlserver.sqlserver_connections import ( + SQLServerConnectionManager, + _discard_pending_results, +) from dbt.adapters.sqlserver.sqlserver_mask import ColumnMask from dbt.adapters.sqlserver.sqlserver_mask import mask_changes as _mask_changes from dbt.adapters.sqlserver.sqlserver_mask import resolve_masks as _resolve_masks from dbt.adapters.sqlserver.sqlserver_relation import SQLServerRelation +from dbt.adapters.sqlserver.sqlserver_runtime import _get_pyodbc logger = AdapterLogger("SQLServer") +# Mirrors sqlserver__select_starts_with_cte +# (dbt/include/sqlserver/macros/adapters/columns.sql): a query opening with a +# CTE cannot be neutered as ``select * from (...) where 1 = 0``, so it reaches +# get_column_schema_from_query unwrapped and would otherwise be executed in +# full just to read its column names. +_SQL_COMMENT = re.compile(r"(?s)/\*.*?\*/|--[^\n]*\n") + +# sp_describe_first_result_set reports true SQL Server types; reading +# ``cursor.description`` reports Python classes, which collapse whole families +# (every integer width arrives as ``int``, every string type as ``varchar``). +# Contract comparison comes through this method either way, so the describe +# path is mapped back onto exactly the names the execute path yields via +# ``data_type_code_to_name``. TestCteProbeAvoidsExecution pins the two +# together; a type missing from this map falls back to executing rather than +# guessing. +# +# The names below are pyodbc's. The class a driver picks for a column is its +# own choice, not SQL Server's, and mssql-python decodes three of these +# differently -- see ``_MSSQL_PYTHON_TYPE_OVERRIDES``. +_SYSTEM_TYPE_TO_EXECUTED_NAME = { + "bigint": "int", + "int": "int", + "smallint": "int", + "tinyint": "int", + "bit": "bit", + "decimal": "decimal", + "numeric": "decimal", + "money": "decimal", + "smallmoney": "decimal", + "float": "float", + "real": "float", + "date": "date", + "time": "time", + "datetime": "datetime2(6)", + "smalldatetime": "datetime2(6)", + "datetime2": "datetime2(6)", + "char": "varchar", + "nchar": "varchar", + "varchar": "varchar", + "nvarchar": "varchar", + "text": "varchar", + "ntext": "varchar", + "xml": "varchar", + "uniqueidentifier": "varchar", + "binary": "varbinary", + "varbinary": "varbinary", + "image": "varbinary", + "timestamp": "varbinary", + "rowversion": "varbinary", + # datetimeoffset is deliberately absent *from this map*: add_query + # registers the -155 output converter after the execute that needed it, + # so pyodbc reports the column as bytearray on a connection's first query + # and as str on every one after. No fixed name mirrors that, so on pyodbc + # describing gives up and the query is executed -- which agrees with + # itself by construction. mssql-python needs no converter and is pinned + # in _MSSQL_PYTHON_TYPE_OVERRIDES. + "sql_variant": "varbinary", + "hierarchyid": "varbinary", + "geography": "varbinary", + "geometry": "varbinary", +} + +# mssql-python decodes three types into richer Python objects than pyodbc +# does: uniqueidentifier as ``uuid.UUID`` (pyodbc: ``str``), datetimeoffset +# as ``datetime`` and sql_variant as ``str`` (pyodbc: ``bytearray`` for +# both). Executing reports those classes, so describing has to agree. +_MSSQL_PYTHON_TYPE_OVERRIDES = { + "uniqueidentifier": "uniqueidentifier", + "datetimeoffset": "datetime2(6)", + "sql_variant": "varchar", +} + + +def _executed_name_for_system_type(base_type: str, backend: Any) -> Optional[str]: + """The name the executed probe would report for a described column type. + + None means "no confident answer" -- the caller then executes the query, + which is slower but cannot disagree with itself. + """ + if is_mssql_python_backend(backend): + overridden = _MSSQL_PYTHON_TYPE_OVERRIDES.get(base_type) + if overridden is not None: + return overridden + + elif base_type == "uniqueidentifier": + # pyodbc yields uuid.UUID or str for a GUID depending on its + # module-level ``native_uuid`` flag -- process-global state anything + # in the process can flip, so read it rather than assume a default. + try: + native_uuid = bool(_get_pyodbc().native_uuid) + except Exception as e: # pragma: no cover - pyodbc is present if in use + logger.debug(f"Could not read pyodbc.native_uuid, executing the query: {e}") + return None + return "uniqueidentifier" if native_uuid else "varchar" + + return _SYSTEM_TYPE_TO_EXECUTED_NAME.get(base_type) + class SQLServerAdapter(SQLAdapter): """ @@ -135,18 +238,108 @@ def _behavior_flags(self) -> List[BehaviorFlag]: @available.parse(lambda *a, **k: []) def get_column_schema_from_query(self, sql: str) -> List[BaseColumn]: - """Get a list of the Columns with names and data types from the given sql.""" + """Get a list of the Columns with names and data types from the given sql. + + Only the result *shape* is wanted, but the query still runs, so the + cursor comes back holding the whole result set. Usually that set is + empty: dbt-core's ``get_column_schema_from_query`` macro wraps the + query first, and ``sqlserver__get_empty_subquery_sql`` renders that as + ``select * from (...) where 1 = 0``. A query that opens with a CTE + cannot be wrapped that way, though, and is passed through untouched + (dbt/include/sqlserver/macros/adapters/columns.sql), so snapshot + staging queries and CTE-headed contract models arrive here in full. + + Either way the cursor must not be abandoned holding rows -- see + ``_discard_pending_results`` for what that costs. + """ + if _SQL_COMMENT.sub("", sql).strip().lower().startswith("with"): + described = self._describe_result_set(sql) + if described is not None: + return described + _, cursor = self.connections.add_select_query(sql) - columns = [ - self.Column.create( - column_name, self.connections.data_type_code_to_name(column_type_code) - ) - # https://peps.python.org/pep-0249/#description - for column_name, column_type_code, *_ in cursor.description - ] + try: + columns = [ + self.Column.create( + column_name, self.connections.data_type_code_to_name(column_type_code) + ) + # https://peps.python.org/pep-0249/#description + for column_name, column_type_code, *_ in cursor.description + ] + finally: + _discard_pending_results(cursor) + return columns + def _describe_result_set(self, sql: str) -> Optional[List[BaseColumn]]: + """Read a query's column shape without running it, or None to fall back. + + ``sp_describe_first_result_set`` compiles the query and reports its + result shape, which is all this method ever wanted. It is already how + ``sqlserver__get_columns_in_query`` handles CTEs (#698). + + Returns None -- deliberately, rather than raising -- whenever the + describe cannot be trusted to match what executing would have reported: + a query it refuses to describe (it cannot see through ``#temp`` tables, + where executing works), or a type this backend's driver has no known + executed name for. The caller then executes as before, which is + slower but never disagrees with itself. + """ + credentials = self.connections.profile.credentials + + # Inline rather than bound: mssql-python binds str as varchar and the + # procedure demands nvarchar(max). columns.sql:24 escapes it the same + # way for the same reason. + describe_sql = "exec sp_describe_first_result_set @tsql = N'{}'".format( + sql.replace("'", "''") + ) + + try: + _, cursor = self.connections.add_select_query(describe_sql) + except Exception as e: + logger.debug(f"Could not describe a CTE query, falling back to executing it: {e}") + return None + + try: + fields = [description[0].lower() for description in cursor.description] + rows = cursor.fetchall() + except Exception as e: + logger.debug(f"Could not read a described result set, executing the query: {e}") + return None + finally: + _discard_pending_results(cursor) + + try: + hidden, name, type_name = ( + fields.index("is_hidden"), + fields.index("name"), + fields.index("system_type_name"), + ) + except ValueError: # pragma: no cover - shape is fixed by SQL Server + return None + + columns = [] + for row in rows: + if row[hidden]: + continue + # "varchar(10)" / "decimal(10,2)" -> "varchar" / "decimal" + base_type = str(row[type_name]).split("(")[0].strip().lower() + executed_name = _executed_name_for_system_type(base_type, credentials.backend) + if executed_name is None or row[name] is None: + logger.debug( + f"Describing a CTE query reported {base_type!r}, which has no " + "equivalent in the executed path; executing it instead" + ) + return None + columns.append(self.Column.create(row[name], executed_name)) + + # Every select has at least one column, so nothing described means + # sp_describe_first_result_set could not work the shape out. Returning + # an empty list would read as "this query has no columns" and surface + # as a baffling contract mismatch; execute instead. + return columns or None + @classmethod def convert_boolean_type(cls, agate_table, col_idx): return "bit" @@ -324,7 +517,10 @@ def _get_row_count(self, relation) -> int: """Return the number of rows in the given relation.""" sql = f"SELECT COUNT_BIG(*) FROM {relation}" _, cursor = self.connections.add_select_query(sql) - row = cursor.fetchone() + try: + row = cursor.fetchone() + finally: + _discard_pending_results(cursor) return int(row[0]) if row else 0 def expand_column_types(self, goal, current, max_rows: int = 1000000): diff --git a/dbt/adapters/sqlserver/sqlserver_connections.py b/dbt/adapters/sqlserver/sqlserver_connections.py index 8e945c1e..678e22d5 100644 --- a/dbt/adapters/sqlserver/sqlserver_connections.py +++ b/dbt/adapters/sqlserver/sqlserver_connections.py @@ -61,6 +61,64 @@ logger = AdapterLogger("sqlserver") + +def _try_drain_nextset(cursor: Any) -> bool: + """Drain one additional result set from *cursor*, if supported. + + Returns ``True`` if a further result set was consumed, ``False`` if the + cursor has no more result sets. + """ + return bool(cursor.nextset()) + + +# Rows per round trip when shedding a result set nobody asked for. Large +# enough that discarding even a big one costs a handful of round trips. +_DISCARD_CHUNK_SIZE = 10000 + + +def _discard_pending_results(cursor: Any) -> None: + """Consume and close *cursor*, leaving nothing for the driver to cancel. + + Closing a cursor whose result set the server is still producing makes the + driver cancel the request, and the cancel reaches SQL Server as an + *attention*. Every connection opened here runs ``SET XACT_ABORT ON`` + (``_apply_session_settings``, for dbt-msft/dbt-sqlserver#718), and SQL + Server answers an attention under ``XACT_ABORT ON`` by rolling back the + open transaction. + + None of that raises -- an attention is not an error -- so a materialization + that abandons a cursor part-way through a build silently loses whatever it + had already created inside that transaction, then fails further on against + relations that no longer exist. Fetching the rows first makes the close an + ordinary one. + + Failures while discarding are logged and swallowed: the caller already has + what it came for, and the connection is about to be reused for real work, + so trouble shedding rows nobody wanted must not become the error the user + sees. + """ + try: + while True: + # ``description`` is None for statements that return no rows at + # all, where fetching would raise rather than yield nothing. + if cursor.description is not None: + while cursor.fetchmany(_DISCARD_CHUNK_SIZE): + pass + # nextset() only advances; whatever rows the set it lands on holds + # still have to be fetched, hence the outer loop. + if not _try_drain_nextset(cursor): + break + except Exception as e: + # AdapterLogger cannot serialize an exception as a log argument, so + # interpolate rather than passing ``e`` through. + logger.debug(f"Discarding a pending result set failed: {e}") + + try: + cursor.close() + except Exception as e: + logger.debug(f"Closing a cursor failed: {e}") + + # Attribute used to stash the in-flight pyodbc / mssql-python cursor on a # Connection so cancel() can reach it from another thread. See cancel(). _IN_FLIGHT_CURSOR_ATTR = "_dbt_sqlserver_in_flight_cursor" diff --git a/dbt/adapters/sqlserver/sqlserver_runtime.py b/dbt/adapters/sqlserver/sqlserver_runtime.py index 414a9778..9ed741af 100644 --- a/dbt/adapters/sqlserver/sqlserver_runtime.py +++ b/dbt/adapters/sqlserver/sqlserver_runtime.py @@ -45,6 +45,11 @@ class PyodbcModuleProtocol(Protocol): InterfaceError: type[Exception] DatabaseError: type[Exception] pooling: bool + # Whether pyodbc decodes a uniqueidentifier column as uuid.UUID rather + # than str. Read by the CTE probe, which has to report the same type + # executing the query would have. Module-global, so it can change under + # us; nothing here sets it. + native_uuid: bool def connect(self, *args: Any, **kwargs: Any) -> Any: ... diff --git a/dbt/include/sqlserver/macros/materializations/models/view/view.sql b/dbt/include/sqlserver/macros/materializations/models/view/view.sql index 3eea0c2d..29ede4a9 100644 --- a/dbt/include/sqlserver/macros/materializations/models/view/view.sql +++ b/dbt/include/sqlserver/macros/materializations/models/view/view.sql @@ -43,10 +43,30 @@ {% elif existing_relation is not none and existing_relation.type == 'view' %} {% set current_view_definition_table = run_query(get_view_definition_sql(existing_relation)) %} {% if current_view_definition_table is not none and current_view_definition_table.rows | length > 0 %} - {% set normalized_relation = target_relation.include(database=False) | lower | replace('\n', '') | replace('\r', '') | replace('\t', '') | replace(' ', '') | replace(';', '') %} - {% set normalized_sql = sql | lower | replace('\n', '') | replace('\r', '') | replace('\t', '') | replace(' ', '') | replace(';', '') %} - {% set normalized_definition = current_view_definition_table.rows[0][0] | lower | replace('\n', '') | replace('\r', '') | replace('\t', '') | replace(' ', '') | replace(';', '') %} - {% set should_skip_view_update = normalized_definition.endswith(normalized_sql) %} + {#- Compare the view *body* exactly, not by suffix. The stored definition is + the whole statement (CREATE [OR ALTER] VIEW AS ); the model is + only the body. The header ends at the separating ' AS ' - split there and + compare the remainder verbatim. A suffix test (endswith) would wrongly skip + any edit whose new body is a tail of the old one, e.g. deleting a leading + comment or CTE - it lands as PASS but never reaches the database, and + --full-refresh does not fix it. Do NOT lowercase or strip whitespace: both + make genuinely different bodies compare equal (a string literal differing + only in case, or any literal containing spaces). The asymmetry is deliberate - a skip that + fails to fire costs one rebuild; a skip that fires wrongly costs correctness - + so when we cannot be certain, we rebuild. -#} + {% set stored = current_view_definition_table.rows[0][0] %} + {#- First ' as ' is the header/body separator: CREATE [OR ALTER] VIEW AS has no other, the relation being quoted. -#} + {% set marker = (stored | lower).find(' as ') %} + {% if marker < 0 %} + {% set should_skip_view_update = false %} + {% else %} + {% set stored_body = stored[marker + 4:] | replace('\r\n', '\n') | trim %} + {% set stored_body = (stored_body[:-1] if stored_body.endswith(';') else stored_body) | trim %} + {% set model_body = sql | replace('\r\n', '\n') | trim %} + {% set model_body = (model_body[:-1] if model_body.endswith(';') else model_body) | trim %} + {% set should_skip_view_update = stored_body == model_body %} + {% endif %} {% endif %} {% if should_skip_view_update %} {% set build_sql = 'declare @dbt_sqlserver_noop int;' %} diff --git a/tests/functional/adapter/dbt/test_constraints.py b/tests/functional/adapter/dbt/test_constraints.py index 328c97cb..ec44c1bb 100644 --- a/tests/functional/adapter/dbt/test_constraints.py +++ b/tests/functional/adapter/dbt/test_constraints.py @@ -750,3 +750,156 @@ def test__constraints_enforcement_rollback( # Its result includes the expected error messages self.assert_expected_error_messages(failing_results[0].message, expected_error_messages) + + +# --------------------------------------------------------------------------- +# Contract enforcement on a model whose SQL opens with a CTE +# --------------------------------------------------------------------------- +# +# Every fixture above renders to a plain `select`, so +# sqlserver__get_empty_subquery_sql wraps it as `select * from (...) where 1 = 0` +# and the contract probe in columns_spec_ddl.sql costs nothing. That wrapper +# cannot wrap a query that already starts with a CTE, and passes it through +# untouched instead (dbt/include/sqlserver/macros/adapters/columns.sql), so a +# CTE-headed model runs in full just to have its column shape read -- and the +# cursor holding the result is abandoned. +# +# Abandoning it while the server is still working on the request makes the +# driver cancel it, and the attention that sends rolls back the open +# transaction under `SET XACT_ABORT ON` without raising anything. The model's +# in-transaction pre-hook is what that rollback destroys here. Sized megabytes +# past the point where the server has finished streaming; see the note in +# tests/functional/adapter/dbt/test_transactions.py for why there is no +# threshold constant to use instead. +_CTE_CONTRACT_ROWS = 5000 + +cte_contract_model_sql = ( + """ +{{ config( + materialized='table', + contract={'enforced': true}, + pre_hook="INSERT INTO {{ this.schema }}.contract_audit_log (msg) VALUES ('before_main')" +) }} +with source_data as ( + select top (%d) + row_number() over (order by (select null)) as id, + cast(replicate('x', 500) as varchar(8000)) as payload + from sys.all_objects a cross join sys.all_objects b +) +select id, payload from source_data +""" + % _CTE_CONTRACT_ROWS +) + +cte_contract_schema_yml = """ +version: 2 +models: + - name: cte_contract_model + config: + contract: + enforced: true + columns: + - name: id + data_type: bigint + - name: payload + data_type: varchar(8000) +""" + + +class TestCteModelConstraintsColumnsEqual: + """The contract probe must not disturb the transaction it runs inside.""" + + @pytest.fixture(scope="class") + def models(self): + return { + "cte_contract_model.sql": cte_contract_model_sql, + "constraints_schema.yml": cte_contract_schema_yml, + } + + def test_contract_probe_leaves_the_transaction_intact(self, project): + project.run_sql( + "CREATE TABLE {schema}.contract_audit_log (msg varchar(100))", + ) + + results = run_dbt(["run", "-s", "cte_contract_model"]) + assert len(results) == 1 + assert results[0].status == "success" + + rows = project.run_sql("select count(*) from {schema}.contract_audit_log", fetch="one") + assert rows[0] == 1, ( + "the pre-hook ran inside the model's transaction and its row is gone: " + "the contract probe abandoned its cursor, the driver cancelled the " + "request, and XACT_ABORT rolled the transaction back" + ) + + relation = relation_from_name(project.adapter, "cte_contract_model") + built = project.run_sql(f"select count(*) from {relation}", fetch="one") + assert built[0] == _CTE_CONTRACT_ROWS + + +# The type set the probe has to report consistently. Contract comparison reads +# these, so whichever way the probe learns a query's shape must agree with the +# other -- see TestCteProbeAvoidsExecution. +_PROBE_TYPE_MATRIX = [ + "bigint", + "int", + "smallint", + "tinyint", + "bit", + "decimal(10,2)", + "numeric(5,1)", + "money", + "float", + "real", + "date", + "time", + "datetime", + "datetime2(3)", + "char(5)", + "nchar(5)", + "varchar(10)", + "nvarchar(20)", + "varchar(max)", + "nvarchar(max)", + "uniqueidentifier", + "varbinary(10)", +] + + +class TestCteProbeAvoidsExecution: + """A CTE-headed query reaches get_column_schema_from_query unwrapped, + because sqlserver__get_empty_subquery_sql cannot neuter it with + `where 1 = 0`. Reading its shape should not mean running it.""" + + def test_cte_probe_does_not_execute_the_query(self, project): + # 1/0 compiles cleanly and only fails when the query actually runs, so + # getting columns back is proof the probe did not execute it. + sql = "with q as (select 1/0 as boom, cast('x' as varchar(10)) as t) select * from q" + + with project.adapter.connection_named("_probe"): + columns = project.adapter.get_column_schema_from_query(sql) + + assert [c.column for c in columns] == ["boom", "t"] + + def test_cte_probe_reports_the_same_types_as_the_wrapped_probe(self, project): + """Guard, not a symptom: this has to hold before and after any change + to how the CTE branch reads metadata, or contract comparisons shift + under models that merely happen to open with a CTE.""" + mismatches = [] + + with project.adapter.connection_named("_probe"): + for type_sql in _PROBE_TYPE_MATRIX: + plain = f"select cast(null as {type_sql}) as c" + wrapped = project.adapter.get_column_schema_from_query( + f"select * from ({plain}) dbt_sbq_tmp where 1 = 0" + ) + cte = project.adapter.get_column_schema_from_query( + f"with q as ({plain}) select * from q" + ) + if [(c.column, c.dtype) for c in cte] != [(c.column, c.dtype) for c in wrapped]: + mismatches.append( + f"{type_sql}: cte={[(c.column, c.dtype) for c in cte]} " + f"wrapped={[(c.column, c.dtype) for c in wrapped]}" + ) + + assert not mismatches, "probe branches disagree on:\n" + "\n".join(mismatches) diff --git a/tests/functional/adapter/dbt/test_transactions.py b/tests/functional/adapter/dbt/test_transactions.py index 9c93d33b..f64728a0 100644 --- a/tests/functional/adapter/dbt/test_transactions.py +++ b/tests/functional/adapter/dbt/test_transactions.py @@ -29,10 +29,33 @@ def project_config_update(self): select 1/0 as boom """ -_snapshot_seed_csv = """id,name,updated_at -1,alice,2024-01-01 00:00:00 -2,bob,2024-01-01 00:00:00 -""" +# The snapshot source is sized deliberately. A snapshot's second run probes the +# shape of its staging query through get_column_schema_from_query +# (check_time_data_types -> get_updated_at_column_data_type), and that query +# starts with a CTE, so sqlserver__get_empty_subquery_sql cannot wrap it in +# `where 1 = 0` and it runs in full. If the cursor holding that result set is +# abandoned while the server is still working on the request, the driver +# cancels it; the attention that sends rolls back the open transaction under +# `SET XACT_ABORT ON`, silently, taking the staging table with it. +# +# The failure is a race rather than a size threshold -- measured against SQL +# Server 2022, a few hundred rows is already enough at zero client delay, while +# ~10ms of client-side work before the close makes even 20MB safe -- so there is +# no constant to encode. These fixtures sit megabytes past the boundary so the +# server is unambiguously still streaming, whatever the runner's speed. A +# handful of narrow rows (which is what a seed gives you) never trips it, which +# is why this went unnoticed. +_SNAPSHOT_ROWS = 5000 +_SNAPSHOT_PAYLOAD_WIDTH = 500 + +_snapshot_source_sql = """ +{{ config(materialized='table') }} +select top (%d) + row_number() over (order by (select null)) as id, + cast(replicate('{{ var("payload_char", "x") }}', %d) as varchar(8000)) as payload, + cast('{{ var("snap_updated_at", "2024-01-01") }}' as datetime2) as updated_at +from sys.all_objects a cross join sys.all_objects b +""" % (_SNAPSHOT_ROWS, _SNAPSHOT_PAYLOAD_WIDTH) _snapshot_sql = """ {% snapshot snap %} @@ -42,7 +65,19 @@ def project_config_update(self): strategy='timestamp', updated_at='updated_at', ) }} -select * from {{ ref('snap_seed') }} +select * from {{ ref('snap_source') }} +{% endsnapshot %} +""" + +_snapshot_check_sql = """ +{% snapshot snap_check %} +{{ config( + target_schema=schema, + unique_key='id', + strategy='check', + check_cols=['payload'], +) }} +select * from {{ ref('snap_source') }} {% endsnapshot %} """ @@ -73,15 +108,12 @@ def models(self): select 1/0 as boom """, "failing_model.sql": _failing_model_sql, + "snap_source.sql": _snapshot_source_sql, } - @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} + return {"snap.sql": _snapshot_sql, "snap_check.sql": _snapshot_check_sql} def test_table_materialization(self, project): results = run_dbt(["run", "--models", "table_model"]) @@ -137,18 +169,50 @@ def test_side_effect_rolled_back(self, project): assert rows[0] == 0 def test_snapshot_create_and_merge(self, project): - run_dbt(["seed"]) + """Timestamp strategy, with a second run that has changes to write. + + The merge reads the staging table built earlier in the same + transaction, so a probe that silently rolls that transaction back + surfaces here as `Invalid object name '..._dbt_tmp'`. + """ + run_dbt(["run", "--models", "snap_source"]) results = run_dbt(["snapshot", "--select", "snap"]) assert len(results) == 1 assert results[0].status == "success" rows = project.run_sql("select count(*) from {schema}.snap", fetch="one") - assert rows[0] == 2 + assert rows[0] == _SNAPSHOT_ROWS + # move every row's updated_at forward, so the second run has a full + # changeset to stage and merge rather than converging to zero rows + run_dbt(["run", "--models", "snap_source", "--vars", "snap_updated_at: '2024-06-01'"]) results = run_dbt(["snapshot", "--select", "snap"]) assert len(results) == 1 assert results[0].status == "success" + rows = project.run_sql("select count(*) from {schema}.snap", fetch="one") + assert rows[0] == _SNAPSHOT_ROWS * 2 + + def test_snapshot_check_strategy_create_and_merge(self, project): + """Same path as above via the check strategy: both strategies build + their staging query through sqlserver__snapshot_staging_table, so both + hand the probe a CTE-headed query that runs in full.""" + run_dbt(["run", "--models", "snap_source"]) + results = run_dbt(["snapshot", "--select", "snap_check"]) + assert len(results) == 1 + assert results[0].status == "success" + + rows = project.run_sql("select count(*) from {schema}.snap_check", fetch="one") + assert rows[0] == _SNAPSHOT_ROWS + + run_dbt(["run", "--models", "snap_source", "--vars", "payload_char: y"]) + results = run_dbt(["snapshot", "--select", "snap_check"]) + assert len(results) == 1 + assert results[0].status == "success" + + rows = project.run_sql("select count(*) from {schema}.snap_check", fetch="one") + assert rows[0] == _SNAPSHOT_ROWS * 2 + class BaseFailingModelWithSideEffect: @pytest.fixture(scope="class") diff --git a/tests/functional/adapter/dbt/test_unit_tests.py b/tests/functional/adapter/dbt/test_unit_tests.py index 8567c800..f6d902eb 100644 --- a/tests/functional/adapter/dbt/test_unit_tests.py +++ b/tests/functional/adapter/dbt/test_unit_tests.py @@ -127,3 +127,55 @@ def test_unit_test_data_type(self, project, data_types): run_dbt(["test", "--select", "my_model"]) except Exception: raise AssertionError(f"unit test failed when testing model with {sql_value}") + + +# The contract branch of sqlserver__unit_test_create_table_as had no coverage. +# It reaches get_assert_columns_equivalent, and so the same probe that runs a +# CTE-headed query in full for snapshots and contract-enforced models (see +# tests/functional/adapter/dbt/test_constraints.py). It is safe here only +# because a unit test's inputs are replaced by its fixture rows, so the query +# being probed returns a handful of rows however large the real model is -- +# this pins that branch so a change to it does not go unnoticed. +contract_unit_test_model_sql = """ +select tested_column from {{ ref('my_upstream_model') }} +""" + +contract_unit_test_yml = """ +version: 2 +models: + - name: my_contract_model + config: + contract: + enforced: true + columns: + - name: tested_column + data_type: int +unit_tests: + - name: test_my_contract_model + model: my_contract_model + given: + - input: ref('my_upstream_model') + rows: + - {tested_column: 1} + expect: + rows: + - {tested_column: 1} +""" + + +class TestUnitTestWithContract: + @pytest.fixture(scope="class") + def models(self): + return { + "my_upstream_model.sql": upstream_model_sql, + "my_contract_model.sql": contract_unit_test_model_sql, + "schema.yml": contract_unit_test_yml, + } + + def test_unit_test_runs_under_an_enforced_contract(self, project): + results = run_dbt(["run"]) + assert len(results) == 2 + + results = run_dbt(["test", "--select", "my_contract_model"]) + assert len(results) == 1 + assert results[0].status == "pass" diff --git a/tests/functional/adapter/mssql/test_materialize_change.py b/tests/functional/adapter/mssql/test_materialize_change.py index 5137d8d1..4db4516a 100644 --- a/tests/functional/adapter/mssql/test_materialize_change.py +++ b/tests/functional/adapter/mssql/test_materialize_change.py @@ -1,6 +1,6 @@ import pytest -from dbt.tests.util import get_connection, run_dbt +from dbt.tests.util import get_connection, run_dbt, write_file model_sql = """ SELECT 1 AS data @@ -33,6 +33,33 @@ SELECT * FROM missing_relation """ +# Same body with and without a leading comment. Removing the comment leaves the +# new body as a *suffix* of the stored definition - the case the old endswith() +# skip test got wrong, skipping the rebuild so the change never reached the db. +view_with_leading_comment = """ +{{ config(materialized='view') }} +-- leading_marker_comment +SELECT 1 AS data +""" + +view_without_leading_comment = """ +{{ config(materialized='view') }} +SELECT 1 AS data +""" + +# Two bodies that differ only by the case of a string literal. Lowercasing before +# comparing (as the old code did) would treat these as identical and skip the +# rebuild - a correctness bug, not just a missed comment. +view_literal_upper = """ +{{ config(materialized='view') }} +SELECT 'ABC' AS source +""" + +view_literal_lower = """ +{{ config(materialized='view') }} +SELECT 'abc' AS source +""" + schema = """ version: 2 models: @@ -161,3 +188,66 @@ def models(self): def test_passes(self, project): self.create_object(project, f"CREATE VIEW {project.test_schema}.mat_object AS {model_sql}") run_dbt(["run"]) + + +def _stored_view_definition(project): + """The whole stored CREATE ... VIEW ... AS statement, as SQL Server keeps it.""" + return project.run_sql( + f"select object_definition(object_id('{project.test_schema}.mat_object'))", + fetch="one", + )[0] + + +class TestViewLeadingTextRemovalReachesDatabase(BaseTableView): + """Removing text from the *start* of a view body must rebuild the view. + + The old skip test compared with ``normalized_definition.endswith(normalized_sql)``. + The stored definition is the whole statement while the model is only the body, + so any edit whose new body is a tail of the old one (e.g. deleting a leading + comment) satisfied endswith() and was silently skipped - PASS, but the change + never reached the database, and --full-refresh did not fix it. + """ + + @pytest.fixture(scope="class") + def models(self): + return {"mat_object.sql": view_with_leading_comment, "schema.yml": schema} + + def test_removal_of_leading_comment_lands(self, project): + run_dbt(["run"]) + assert "leading_marker_comment" in _stored_view_definition(project) + + # Delete the leading comment - the new body is now a suffix of the old. + write_file(view_without_leading_comment, "models", "mat_object.sql") + results = run_dbt(["run"]) + assert len(results) == 1 + + assert "leading_marker_comment" not in _stored_view_definition(project) + + +class TestViewLiteralCaseChangeRebuilds(BaseTableView): + """A change confined to the case of a string literal must rebuild the view. + + The old skip test lowercased both sides before comparing, so ``'ABC'`` and + ``'abc'`` looked identical and the rebuild was skipped - a correctness bug, + since the two views return different data. The exact comparison rebuilds. + """ + + @pytest.fixture(scope="class") + def models(self): + return {"mat_object.sql": view_literal_upper, "schema.yml": schema} + + def test_case_only_change_lands(self, project): + run_dbt(["run"]) + assert ( + project.run_sql(f"select source from {project.test_schema}.mat_object", fetch="one")[0] + == "ABC" + ) + + write_file(view_literal_lower, "models", "mat_object.sql") + results = run_dbt(["run"]) + assert len(results) == 1 + + assert ( + project.run_sql(f"select source from {project.test_schema}.mat_object", fetch="one")[0] + == "abc" + ) diff --git a/tests/unit/adapters/mssql/test_expand_column_types.py b/tests/unit/adapters/mssql/test_expand_column_types.py index 3aafc53b..4cd33439 100644 --- a/tests/unit/adapters/mssql/test_expand_column_types.py +++ b/tests/unit/adapters/mssql/test_expand_column_types.py @@ -166,3 +166,45 @@ def test_varchar_max_to_bounded_does_not_expand(self, adapter): adapter.expand_column_types(goal, current, max_rows=-1) adapter.alter_column_type.assert_not_called() + + +class TestGetRowCount: + """expand_column_types' row-count probe owns the cursor it is handed by + add_select_query, and has to release it.""" + + @pytest.fixture + def raw_adapter(self): + config = MagicMock() + config.flags = {} + config.project_name = "test" + config.credentials.type = "sqlserver" + return SQLServerAdapter(config, MagicMock()) + + @staticmethod + def _cursor(count=42): + """A single-row COUNT_BIG result. Explicit about being exhausted -- + a MagicMock's fetchmany/nextset are truthy forever, which would hang + any caller that drains before closing.""" + cursor = MagicMock() + cursor.fetchone.return_value = (count,) + cursor.fetchmany.return_value = [] + cursor.nextset.return_value = False + return cursor + + @staticmethod + def _attach(adapter, cursor): + adapter.connections.add_select_query = MagicMock(return_value=(MagicMock(), cursor)) + return cursor + + def test_returns_the_count(self, raw_adapter): + self._attach(raw_adapter, self._cursor()) + + assert raw_adapter._get_row_count(make_rel()) == 42 + + def test_closes_the_cursor(self, raw_adapter): + cursor = self._cursor() + self._attach(raw_adapter, cursor) + + raw_adapter._get_row_count(make_rel()) + + cursor.close.assert_called_once() diff --git a/tests/unit/adapters/mssql/test_get_column_schema_from_query.py b/tests/unit/adapters/mssql/test_get_column_schema_from_query.py new file mode 100644 index 00000000..05bdd26a --- /dev/null +++ b/tests/unit/adapters/mssql/test_get_column_schema_from_query.py @@ -0,0 +1,138 @@ +"""A cursor taken from ``add_select_query`` must be drained and closed. + +Abandoning a cursor while the server is still streaming its result makes the +driver cancel the request. The cancel arrives as an *attention*, and every +connection this adapter opens runs under ``SET XACT_ABORT ON`` +(``SQLServerConnectionManager._set_session_options``, #718), so SQL Server +answers an attention by rolling back the open transaction. An attention is not +an error, so nothing is raised and nothing is logged: the caller carries on and +fails later against relations the rollback removed. + +These tests pin the invariant without a warehouse -- the cursor must be closed, +and it must have nothing left outstanding at the moment it is closed. The +end-to-end versions live with the callers that trip it: the snapshot staging +probe in tests/functional/adapter/dbt/test_transactions.py and the contract +probe in tests/functional/adapter/dbt/test_constraints.py. +""" + +from unittest.mock import MagicMock + +import pytest + +from dbt.adapters.sqlserver.sqlserver_adapter import SQLServerAdapter + + +class FakeCursor: + """Records whether anything was still outstanding when it was closed.""" + + def __init__(self, rows, description=None, extra_result_sets=0): + self._rows = list(rows) + self._extra_result_sets = extra_result_sets + # PEP 249: (name, type_code, display_size, internal_size, precision, + # scale, null_ok) + self.description = description or [ + ("id", 4, None, None, None, None, None), + ("payload", 12, None, None, None, None, None), + ] + self.closed = False + self.rows_pending_at_close = None + self.result_sets_pending_at_close = None + self.fetchmany_calls = 0 + + def fetchmany(self, size): + self.fetchmany_calls += 1 + batch, self._rows = self._rows[:size], self._rows[size:] + return batch + + def fetchone(self): + return self._rows.pop(0) if self._rows else None + + def nextset(self): + if self._extra_result_sets: + self._extra_result_sets -= 1 + self._rows = [("row",)] * 25 + return True + return False + + def close(self): + self.closed = True + self.rows_pending_at_close = len(self._rows) + self.result_sets_pending_at_close = self._extra_result_sets + + +@pytest.fixture +def adapter(): + config = MagicMock() + config.flags = {} + config.project_name = "test" + config.credentials.type = "sqlserver" + return SQLServerAdapter(config, MagicMock()) + + +def attach(adapter, cursor): + adapter.connections.add_select_query = MagicMock(return_value=(MagicMock(), cursor)) + adapter.connections.data_type_code_to_name = MagicMock(return_value="varchar") + return cursor + + +class TestGetColumnSchemaFromQuery: + def test_returns_the_column_schema(self, adapter): + attach(adapter, FakeCursor(rows=[("a", "b")] * 100)) + + columns = adapter.get_column_schema_from_query("select 1") + + assert [c.column for c in columns] == ["id", "payload"] + + def test_closes_the_cursor(self, adapter): + cursor = attach(adapter, FakeCursor(rows=[("a", "b")] * 100)) + + adapter.get_column_schema_from_query("select 1") + + assert cursor.closed, "cursor was abandoned instead of closed" + + def test_leaves_no_rows_outstanding_at_close(self, adapter): + cursor = attach(adapter, FakeCursor(rows=[("a", "b")] * 25_000)) + + adapter.get_column_schema_from_query("select 1") + + assert cursor.rows_pending_at_close == 0, ( + "closing with rows still pending makes the driver cancel the request, " + "which rolls back the open transaction under XACT_ABORT ON" + ) + + def test_leaves_no_result_sets_outstanding_at_close(self, adapter): + """Draining must re-fetch after each ``nextset()``, not just advance past it.""" + cursor = attach(adapter, FakeCursor(rows=[("a", "b")] * 50, extra_result_sets=2)) + + adapter.get_column_schema_from_query("select 1") + + assert cursor.result_sets_pending_at_close == 0 + assert cursor.rows_pending_at_close == 0 + + def test_fetches_in_batches_rather_than_one_row_at_a_time(self, adapter): + cursor = attach(adapter, FakeCursor(rows=[("a", "b")] * 100_000)) + + adapter.get_column_schema_from_query("select 1") + + # 100k rows in a bounded number of round trips, not 100k of them + assert 0 < cursor.fetchmany_calls <= 50 + + def test_closes_the_cursor_when_column_building_raises(self, adapter): + cursor = attach(adapter, FakeCursor(rows=[("a", "b")] * 100)) + adapter.connections.data_type_code_to_name = MagicMock(side_effect=ValueError("boom")) + + with pytest.raises(ValueError): + adapter.get_column_schema_from_query("select 1") + + assert cursor.closed, "cursor leaked when column building failed" + + def test_a_failure_while_discarding_is_not_raised_to_the_caller(self, adapter): + """The caller already has its metadata; a discard problem must not become + the error the user sees.""" + cursor = attach(adapter, FakeCursor(rows=[("a", "b")] * 100)) + cursor.fetchmany = MagicMock(side_effect=RuntimeError("driver went away")) + + columns = adapter.get_column_schema_from_query("select 1") + + assert [c.column for c in columns] == ["id", "payload"] + assert cursor.closed