backport(1.11): two silent-correctness fixes for v1.11.1

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 <name> 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

dbt/adapters/sqlserver/__version__.py

@@ -1 +1 @@
-version = "1.11.0"
+version = "1.11.1"

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):

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"

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: ...
 

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 <name> AS <body>); 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 <quoted
+          relation> 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;' %}